Google Cloud Dataflow是一种完全托管的云服务，可用于大规模创建和评估数据处理管道。数据流管道基于Apache Beam编程模型，可以在批处理和流模式下运行。 Cloud Dataflow是Google云端平台的一部分。
有什么办法可以在Google Cloud Spanner中删除超过2万个突变？
我在Spanner表中有数百万条记录，我想使用某些查询条件从Spanner删除行。例如：从编号> 2000的扳手表中删除。我无法运行此查询...
具有通用功能的Google Cloud Dataflow自定义键
我们正在使用Dataflow Java SDK，并且越来越多的自定义键类几乎相同。我想让他们扩展一个通用的抽象类，但是Dataflow SDK ...
即使区域设置为bigquery数据集的区域，数据流作业也无法写入不同区域的BigQuery数据集
我们正在编写一个数据流作业，以将存储桶中的JSON数据写入BigQuery数据集。存储桶和BigQuery数据集都在区域X中。但是，数据流终结点在...
以下代码构建了管道，并生成了DAG。 RuntimeError：NotImplementedError [在运行'generatedPtransform-438'时]请让我知道在...
如何在Apache Beam中将DoFn PTransform应用于PCollectionTuple
我正在尝试将PTransform应用于PCollectionTuple，但无法弄清编译器为何抱怨。我想这样做是为了将多个步骤抽象为一个PTransform ...
我正在尝试创建一个Beam输入，该输入每秒执行一次，只是将时间作为输入输出。我知道我可以从类似p.apply（Create.of（1、2、3、4、5））的数字中进行pcollection ...
Python数据流DoFN类函数finish_bundle多次运行并提供空输出
我正在运行一个数据流管道，在该管道中，我必须将数据整理到一个Python数据帧中以用于下一步。因此，我使用DoFn类并定义__init__，process和finish_bundle函数...
我现在正在努力使用管道模板。输入文件是否压缩时，我没有相同的行为。这是一个csv文件的工作日志：拆分文件模式gs：// bucket / file ....
如何保留公共IP（静态IP）以执行Google数据流作业，以便我可以将源应用程序中的IP列入白名单？
我想使用Google数据流作业从本地SQL Server数据库中提取数据，因此想将数据流VM IP列入白名单。要在本地SQL Server上将IP列入白名单，需要一个静态IP。请让...
我已经使用Python SDK（Apache Beam Python 3.7 SDK 2.19.0）构建了一个窗口式流数据流管道。初始数据的表示形式是：电话号码|通话时长| -------------- || ......
如何使用自定义Docker映像运行Python Google Cloud Dataflow作业？
我想使用自定义Docker映像运行Python Google Cloud Dataflow作业。根据文档，这应该是可能的：https：//beam.apache.org/documentation/runtime/environments /＃...
和Beam一起玩之后，我必须在这里问。有没有办法建立按事件时间将项目缓冲X + 2秒的管道？刷新前X秒的数据；转到步骤1；我一直在阅读...
我有一个用例，其中我正在读取数十亿条记录，但是我需要限制记录以查看数据行为。我有一个pardo，我在其中分析有限的数据并执行一些...
运行数据流作业（java）时PubsubIO.readAvroGenericRecords上的空指针异常
我有以下Apache光束管道：软件包ch.mycompany.bb8;导入ch.mycompany.bb8.transforms.LogRecords;导入java.io.File;导入java.io.IOException;导入org.apache.avro.Schema; ...
自定义类型的PCollection汇总的管道性能-属性的均值和中位数
我的Apache-Beam / Dataflow管道通过pub / sub事件构建自定义类型的键控PCollection。现在，预汇总输出为PCollection >。接下来我需要...
我有一个DataFlow管道，该管道使用来自Cloud Pub / Sub的JSON消息，将其转换为Avro并将其存储到Google Cloud Storage。有没有一种方法可以让DataFlow将每个Avro对象存储到...
bigquery数据集的文档指出可以使用以下欧洲地区（A）：europe-north1 europe-west3 europe-west2 europe-west6数据流的文档指出这些是可用的...
更新：“状态”没有被索引:(我索引了该属性，查询运行良好。当使用DirectPipelineRunner运行数据流管道时，我的查询没有任何匹配，尽管我可以看到...
[当即时消息通过以下步骤使用数据流时：-从bigquery中读取-将表行转换为json字符串-插入elasticsearch（7.5.2）它看起来适合〜100k条记录，但实际上（8m条记录〜...] >