无法从Apache beam中的本地仿真器读取发布/订阅消息
Apache Beam是一个开源的分布式数据处理框架，它提供了一种统一的编程模型，可以在不同的批处理和流处理引擎上运行。本地仿真器是Apache Beam提供的一种用于开发和测试的工具，它模拟了分布式环境，但在本地运行，以便开发人员可以更轻松地调试和验证他们的代码。
然而，Apache Beam的本地仿真器目前不支持从发布/订阅消息中读取数据。发布/订阅消息是一种常见的消息传递模式，其中消息发布者将消息发送到一个或多个主题，而订阅者则从这些主题中接收消息。这种模式通常用于构建实时数据流处理系统。
如果您需要从发布/订阅消息中读取数据，并在Apache Beam中进行处理，可以考虑使用其他适用的技术和工具。以下是一些可能的选择：
- 使用Apache Kafka：Apache Kafka是一个分布式流处理平台，它提供了持久化的、高吞吐量的消息队列。您可以使用Apache Kafka作为消息中间件，将消息发送到主题，并使用Apache Beam连接到Kafka主题以读取和处理数据。
- 使用Apache Pulsar：Apache Pulsar是一个分布式的、持久化的消息和流处理平台，它提供了高吞吐量和低延迟的消息传递。您可以使用Apache Pulsar作为消息中间件，将消息发送到主题，并使用Apache Beam连接到Pulsar主题以读取和处理数据。
- 使用其他云原生消息队列服务：根据您的需求，您可以选择使用云厂商提供的消息队列服务，如腾讯云的消息队列CMQ、阿里云的消息队列RocketMQ等。这些服务通常提供了可靠的消息传递和高可用性，您可以使用Apache Beam连接到这些服务以读取和处理数据。
需要注意的是，以上提到的技术和工具仅是一些示例，具体选择应根据您的需求和实际情况来决定。在使用这些技术和工具时，您可以参考官方文档和相关教程，以了解更多详细信息和使用方法。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CMQ：https://cloud.tencent.com/product/cmq
- 腾讯云云原生消息队列 TDMQ：https://cloud.tencent.com/product/tdmq
相关·内容
我正在尝试使用DirectRunner运行一个简单的Apache Beam管道，它从发布/订阅订阅中读取消息并将消息写入磁盘。当我在GCP上运行它时，管道工作得很好，但是当我尝试在我本地的Pub/Sub仿真器上运行它时，它似乎什么都不起作用。我使用了一个扩展org.apache.beam.sdk.io.gcp
浏览 18提问于2019-04-30得票数 0
回答已采纳
1回答
我已经用Apache光束写了一个数据流管道，让你对代码有一个基本的了解： Job= ( |"cretae">>beam.Create(["message"]) |"job 2" >> beam.ParDo(dofn2())
|"job 3" >>
浏览 24提问于2021-09-03得票数 0
回答已采纳
1回答
我有一个云函数，它将消息发布到PubSub，并触发云运行来执行归档文件过程。当有大文件时，我的cloud Run python代码需要一些时间来处理数据，它看起来像是PubSub在20秒(默认确认截止时间)后重试消息，这将从我的Cloud run触发另一个实例。我已经将确认截止时间增加到600秒，并重新部署了所有内容，但它在20秒后仍在重试消息。我遗漏了什么吗？云函数发布消息代码： # Publishes a message
浏览 17提问于2021-07-02得票数 2
回答已采纳
我正在本地计算机上创建一个POC。POC将使用apache beam从订阅中读取Json Pub/Sub消息，对于该订阅，我有一个服务帐户和凭据json。问题是如何通过json文件提供凭据？似乎没有方法，像这样的PipelineOptions接口
setOption("credentials", new Credentials()) ...
浏览 1提问于2019-09-04得票数 0
1回答
我正在尝试将数据从云发布/订阅传递到Google云存储。当我使用runner DataflowRunner时，管道被发布到Google Cloud Dataflow，并按预期工作。然而，对于一些测试，我希望管道在本地运行(但仍然从云发布/订阅读取并写入云存储)。当我使用运行器DirectRunner时，该流程写出INFO:apache_beam.runners.direct.di
浏览 11提问于2021-02-23得票数 0
回答已采纳
我有一个在pubSub中创建主题并将消息发布到该主题的程序。我还有一个自动的数据流作业(使用模板)，它将这些消息保存到我的BigQuery表中。现在我打算用python管道替换基于模板的作业，我的需求是从PubSub读取数据，应用转换，并将数据保存到BigQuery/发布到另一个PubSub主题。我开始用python编写脚本，并做了很多尝试和错误来实现它
浏览 3提问于2019-09-16得票数 0
Google Cloud Platform for Java SDK 2.x的很多文档都告诉您要参考Beam文档。当使用数据流从PubSub读取时，我是否应该仍然执行PubsubIO.Read.named("name").topic("")；
或者我应该做一些其他的事情？在此基础上，有没有一种方法可以将数据流接收到的PubSub数据打印到标准输出或文件中？
浏览 1提问于2018-01-23得票数 2
我正在尝试通过数据流模板"Pub/Sub Avro to Bigquery“将数据从Pub/Sub流式传输到Bigquery。Pub/Sub中的数据是AVRO格式的，来自Kafka主题。我从模式注册表中获得的相应模式文件。java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) java.lang.Thread.run(Threa
浏览 32提问于2021-01-22得票数 0
我有一个发布/订阅主题，它将定期(通常每隔几天或几周一次，但有时更频繁)接收批量消息。我想启动一个批处理数据流作业来读取这些消息，执行一些转换，将结果写入数据存储，然后停止运行。当一批新的消息发出时，我想开始一项新的工作。我已经阅读了Apache Beam Python SDK文档和许多问题，但对一些事情仍然不确定。
是否可以将发布/订阅IO作为非流作业
浏览 2提问于2019-02-21得票数 1
谷歌数据流-如果从数据流写入本地服务器，如何指定TextIO？(前提是本地服务器通过云VPN连接到GCP )在运行数据流作业时，是否可以像上面那样提供本地IP和目录？我试过了，作业成功完成，并在步骤摘要中添加了元素，但我没有看到任何写入<
浏览 2提问于2018-09-14得票数 0
2回答
我正在使用云数据流将数据从发布/订阅消息导入到BigQuery表中。我使用DynamicDestinations，因为这些消息可以放入不同的表中。$DatasetServiceImpl.insertAll(BigQueryServicesImpl.java:765) at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl(StreamingWr
浏览 4提问于2019-02-16得票数 4
我从查找中读取一些客户记录并将其写入bigquery表，然后从该表中读取一些必需的数据字段，并尝试以批处理模式使用数据流管道将该数据(Json)作为消息推送到pubsub中。但出现错误："ValueError: Cloud发布/订阅当前仅可用于流媒体管道“。delete_rows = p | 'reading data to be deleted' >&
浏览 1提问于2019-08-06得票数 0
我在Maven中使用apache apache，在pom.xml中，依赖项是 <groupId>org.apache.beam</groupId> <version>2.8.0</version>
</dependency>
浏览 0提问于2019-01-17得票数 1
我正在开发一个POC，它是方法评估所必需的。我的Mac中安装了python、venv、apache beam和gcloud。我还登录了gcloud傀儡。下面的代码创建订阅我的PubSub并读取来自Pubsub的消息。步骤：但是没有
浏览 5提问于2021-02-08得票数 2
在我的项目中，我希望使用Google Dataflow中的流水线来处理发布/订阅消息。在清理输入数据时，我也希望有一个来自BigQuery的侧输入。这就出现了一个问题，它将导致两个输入中的一个无法工作。 我在我的管道选项中设置了streaming=True，它允许正确处理发布/订阅输入。/订阅当前仅
浏览 29提问于2019-01-10得票数 2
我正在寻找一种方法，使谷歌DataFlow作业停止摄取从发布/订阅时，一个(特定的)异常发生。如果从PubSub接收的事件/行中的字段不是目标BigQuery表中的列，则DataFlow作业会在运行时记录IOExceptions，声明它无法插入行，但似乎会确认这些消息并继续运行。至少，我希望确保那些无法插入到BigQuery中
浏览 5提问于2017-06-10得票数 5
回答已采纳
我已经在Google Cloud上部署了一个9节点的Datastax集群。发布/订阅是否可以将数据加载到Google Cloud上的Datastax<
浏览 1提问于2016-11-17得票数 0
我有一个Pub/Sub脚本，发布男性名字如下：import names
publisher.publish(topic_path, data=data)i
浏览 1提问于2020-03-26得票数 0
回答已采纳
试图做一些概念上简单的事情，但却让我的头撞到了墙上。我尝试在Python语言中创建一个流式数据流作业，它使用来自PubSub主题/订阅的JSON消息，对每个消息执行一些基本操作(在本例中，将温度从C转换为F)，然后将记录发布回不同的主题：import argparse
import apa
浏览 3提问于2018-06-29得票数 1
我在测试google cloud pub/sub的流媒体处理。将消息从发布者转发到主题，在apache-beam上读取发布/订阅上的消息，并使用beam.Map(打印)进行检查。阅读来自pub/sub的消息，它起作用了。但是，在读取所有消息后发生错误。 ㅡ。此代码将消息</
浏览 13提问于2020-08-20得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
