Google Cloud Dataflow是一种完全托管的云服务，可用于大规模创建和评估数据处理管道。数据流管道基于Apache Beam编程模型，可以在批处理和流模式下运行。 Cloud Dataflow是Google云端平台的一部分。
我正在使用Dataflow（Apache beam）创建一个管道来读取和写入Google BigQuery上的数据，但是我在创建DAG方面遇到了问题，就像我对Airflow一样。这是我的代码中的一个例子：＃...
安排Google Cloud Dataflow作业的最简便方法
我只需要每天运行一个数据流管道，但在我看来，建议像App Engine Cron Service这样需要构建整个Web应用程序的解决方案似乎有点太多了。我曾是 ...
我在运行数据流作业时遇到错误。我试图将我现有的光束版本更新为2.11.0，但我在运行时遇到错误。 java.lang.IncompatibleClassChangeError：...
所以我已经阅读了梁的状态处理和及时处理文章，并发现了实现这些功能的问题。我试图解决的问题类似于......
使用--experiments = upload_graph获取Dataflowrunner
我有一个生成数据流图（序列化JSON表示）的管道，它超出了API的允许限制，因此无法通过apache beam的数据流运行器启动...
将此权重/比例转换为列名列表，并使用Python根据其权重/分数矩阵格式进行排序
将此权重/分数读数从输入.csv文件转换为列名列表，并使用Python Apache Beam按其降序权重/分数矩阵格式排序并写入...
这是一个组成的玩具示例，试图获得有关我的问题的更难部分的帮助。假设我有来自Kafka流的销售数据：...期间：5，SalesPersonId：78，销售：TRUE，......
在阅读了使用Apache Beam的Timely（和Stateful）处理并查看了JavaDoc for Timer之后，我成功地设置了一个被触发的计时器。但是，我错过了如何请求计时器......
如何在python梁中制作通用的Protobuf Parser DoFn？
上下文我正在使用流媒体管道，该管道在pubsub中有一个protobuf数据源。我希望将这个protobuf解析为python dict，因为数据接收器要求输入是一个集合......
为什么我的PAssert与我的PCollection中的项目不匹配？
我有一个PCollection，我肯定包含：“Bob”“John”“Fred”但是，当我测试一个断言，询问“Bob”是否在PCollection中使用：PAssert.that（myPcollection）....
我正在通过apache-beam编写一个简单的python管道来聚合用户投票。在输入中，我有这样的逗号分隔行：pollA，answerB pollA，answerC pollB，answerA pollB，answerB pollC，answerE ...
Apache Beam不支持Kotlin Iterable？
Apache光束似乎拒绝承认Kotlin的Iterable。下面是一个示例代码：@ProcessElement fun processElement（@Element input：KV >，接收者：......
我想在事件时间设置一个Timer，它根据我的DoFn中元素中看到的最小时间戳来触发。
我在运行数据流作业时看到了所有类型的奇怪错误（Beam 2.12）。该作业基本上从pubsub获取输入，从/向Datastore读/写将结果写入pubsub。一些 ...
Google Cloud Dataflow Python，检索作业ID
我目前正在使用Python中的数据流模板，我想访问作业ID并使用它来保存到特定的Firestore文档。是否可以访问作业ID？我找不到 ...
我有一个数据流管道，从GCS中的文件读取数据，转换它并将结果放到BQ。我创建了测试，检查预期的TableRows是否在PCollection中，但我怎么能......
导入WriteToDatastore时出错（Apache Beam / Google DataFlow）
我正在尝试使用Apache Beam管道将实体写入Google Cloud Datastore。为了测试，我在使用Apache Beam设置的本地Python 2.7虚拟环境中执行此操作...
在Apache Beam / DataFlow中计算无界流的定期检查点
我正在使用全局无界流与状态处理和计时器相结合，以便按事件时间戳完全按顺序对每个键进行排序。解决方案的答案是......
使用TFRecord和压缩GZIP的Beam Java SDK
我们正在使用Beam Java SDK（以及Google Cloud Dataflow来运行批处理作业），当我们尝试将TFRecordIO与Compression.GZIP一起使用时，我们发现了一些奇怪的东西（可能是一个错误？）。我们有能力 ...
使用CoGroupByKey传输Java SDK 2.4 / 2.5 PAssert
我可能会遗漏一些明显的东西，但由于某些原因我无法使PAssert和TestPipeline与CoGroupByKey一起工作 - 但没有它，它可以正常工作。这是一个参考测试文件，可以......