Apache Beam是用于批处理和流处理的统一SDK。它允许使用特定于Beam的DSL指定大规模数据处理工作流程。 Beam工作流可以在不同的运行时执行，如Apache Flink，Apache Spark或Google Cloud Dataflow（云服务）。
我正在尝试使用Apache Beam建立事件的管道。我想做的是使用事件中的ID从GCP PubSub中读取流数据并从MySQL中读取相关元数据，然后将...
为什么在apache-beam中出现错误：“ TypeError：使用SessionWindow时无法将GlobalWindow转换为_IntervalWindowBase？
[当我使用间隔为1h的会话窗口并处理了数百万条消息后，我在日志中可能会出错，可能仅是针对某些行：TypeError：无法将GlobalWindow转换为apache_beam.utils...。
如何实现Beam sql udf函数将一列拆分为多列？我已经在bigquery udf函数中实现了此功能：CREATE TEMP FUNCTION parseDescription（description STRING）...
我正在尝试使用Gradle在Java中创建数据流。我创建了一个gradle任务，以通过DataflowRunner将作业上传到GCP，并通过环境“ GOOGLE_APPLICATION_CREDENTIALS”设置凭据，...
将CoGbkResult与PAssert.containsInAnyOrder（）结合使用
我正在尝试覆盖一个返回返回接受并返回KV的Apache Beam函数：公共类ChangeKeyFn扩展了DoFn ，KV > {...
我们已经创建了一个非常简单的管道来处理发布子事件。 pub子消息有效负载本身是制表符分隔的csv数据。读取消息后，当...
apap beam 2.19.0不再在云数据流上运行，原因是找不到满足要求setuptools的版本> = 40.8
几天以来，我们的python数据流作业在启动工作程序时导致错误：“错误：找不到满足setuptools要求的版本> = 40.8.0（来自版本：无）”错误：...
如何为Apache Beam中的Avro SpecificRecordBase的所有子类选择单个编码器？
Background My Beam Pipeline设计为处理Avro SpecificRecordBase类型的元素。为了简化我的问题，假设我有两种以Avro格式生成的元素，它们都具有...
我有一个看起来像这样的光束管道p.apply（GenerateSequence.from（0）.withRate（1，new Duration（1000））.apply（ParDo.of（new ReadFromTableFn（config）））;我想通过...
我正在用Java开发Dataflow，输入来自Pubsub。后来，我在这里看到了有关如何使用本地Pubsub仿真器的指南，因此无需进行测试即可部署到GCP。这是我简单的...
到目前为止，在Apache Beam中处理了一个给定HTTP代码的场景，我可能会保留要在下一次迭代中重新启动的元素。一直在用内部代码实现，仅...
我在Google Cloud Dataflow上的Apache Beam管线上遇到麻烦。管道很简单：从GCS读取json，从一些嵌套字段中提取文本，然后写回GCS。它可以工作...
是否可以通过SSH隧道使用Apache Beam jdbcIO？
我需要通过ssh隧道从Mysql服务器中获取数据。我在Google Dataflow上使用Apache Beam 2.19.0 Java JdbcIO连接数据库。但是由于数据库位于私有内部...
我是Cloud Dataflow / Apache Beam的新手，所以概念/编程对我来说仍然很模糊。我想要做的是Dataflow侦听Pubsub并以JSON格式获取此消息：{“ ...
在groupbyKey之后的以下代码中，我正在获取PCollection >>。在发送到FileIO之前如何展平Iterable中的值。 .apply（GroupByKey。 create（））.apply（“ ...
我想下载c4数据集。按照说明页面：https：//www.tensorflow.org/datasets/catalog/c4，建议使用数据流。我按照此处介绍的步骤操作：https：// www ....
上下文大家好，我一直在使用Apache Beam管道来生成要存储在GCS中的列式DB，我有一个来自Kafka的数据流，并且有一个1m的窗口。我想转换...
如何在本地计算机上为python安装apache beam SDK
[我很困惑，因为我试图通过从不同的文档中搜索到的内容来安装适用于python的apache beam SDK，从PiPy下载SDK之后，此命令需要...
此问题与了解在Apache Beam中连接gcp管道的语法更相关。这是我当前管道的设置选项= dataflow_options（project_id = project_id，topic_name = ...