google-cloud-dataflow 相关问题

如何检查哪个用户已停止GCP中的数据流管道？

[我有一条运行在GCP上的数据流管道，该管道从pub / sub中读取消息并写入GCS存储桶。我的数据流管道状态被某些用户取消，并且我想知道该用户是谁？

动态扩展Cloud Dataflow作业

我有一个看起来像这样的Cloud Dataflow管道：从Cloud Bigtable读取进行一些转换写入GCS最初不设置任何最大工作线程，数据流自动缩放将缩放为max（...

使用Apache Beam数据流从CDC记录的PCollection合并填充最新值的单行

更改数据捕获（CDC）记录将不具有记录中各列的所有值。对于一条记录的主键，例如R1，我们可能具有CDC的R1的CDC记录的PCollection ...

Google Dataflow仅为大型.bz2文件创建一个工作线程

我正在尝试使用Cloud Dataflow处理Wikidata json转储。我已经从https://dumps.wikimedia.org/wikidatawiki/entities/latest-all.json.bz2下载该文件，并将其托管到GS中...

Composer看不到数据流作业成功完成

我正在使用Gcloud Composer启动Dataflow作业。我的DAG由两个Dataflow作业组成，应该一个接一个地运行。从airflow.contrib.operators.dataflow_operator导入日期时间...

如果要将bigquery表数据从一个区域复制到另一区域，应该如何以及在何处运行数据流？

我正在尝试将Bigquery表数据从一个区域复制到另一个区域，例如亚洲和欧洲。] >>

使用apache光束读取gzip压缩文件，包裹在TextIOWrapper中，结果“ CompressedFile”对象没有属性“可写”]]

我正在努力在Apache Beam中实现一个简单的CSV读取器，以及来自光束仓库的测试：https：//github.com/apache/beam/blob/b85795adbd22d8b5cf9ebc684ce43e172a789587/sdks/python / ...] >

可拆分DoFn导致随机播放键太大问题

我正在尝试实现ListFlatten函数，我已经使用SimpleDoFn实现了该函数，该函数运行良好，但可以进行并行化。我将功能转换为可拆分功能。我设法...

Apache Beam Wait.on JdbcIO.write具有无限制的PCollection问题

我正在尝试将以下情况与无限制的pCollection数据源（PubSub）结合使用。 https://issues.apache.org/jira/browse/BEAM-6732我能够写到DB1。 DB2正在使用Wait.on DB1（PCollection ....

使用ValueProvider格式化数据流中的BigQuery

我目前正在与Dataflow一起在python中进行循环批处理。基本上，我从bigquery读取数据并对其进行处理。.我的管道看起来像这样pipeline_options = ...

日文字符已损坏，当在Google数据流上运行apache Beam时

我正在google dataflow上运行Apache Beam管道。它从GCS存储桶中读取数据，并在处理后将其写入GCS存储桶。该管道处理日语数据。在堆栈驱动程序日志中，日语...

如何解决“ AttributeError：'str'对象没有属性'items'”，这是从PubSub读取并写入BigQuery的数据流管道中的内容>> [

Beam SQL Not Firing

[我正在构建一个简单的原型，其中我从Pubsub中读取数据，并使用BeamSQL，如下代码段所示val eventStream：SCollection [String] = sc.pubsubSubscription [String]（“ projects / ...

Dataflow Python流式自动缩放

[Google Cloud Dataflow何时将支持Python Streaming管道上的自动缩放？这将帮助我降低成本，并缩短管道响应时间。

如何根据云数据流python sdk中的列值拆分csv文件

我想使用ReadFromText从GCS读取csv文件，并希望根据列值拆分成多个文件。请参见下面的示例数据。Col1 Col2 Col3 Value1数据日期value2 ...

Python数据流作业未能提交

[我们在GCP上有一个kubernetes cron作业，该作业提交同一Python数据流作业的多个副本，每个副本都在其自己的容器中。每当我们需要工作的新副本时，只要将其添加到spec-> ...

基于主键将两个表从BigQuery导出到CSV

我在BigQuery表中有2个表经过汇总和处理，分别可能有多达200万行和1000万行。它们具有非常不同的列，但每个列都相同。...

Dataflow状态处理中的一次语义

我们正在尝试在流式传输环境中涵盖以下场景：自作业开始以来计算用户事件的总计（比方说计数）用户事件的数量是不受限制的（...

在GCP上使用python numpy和pandas部署小批量作业

我有一个日常的小型计算工作，该工作从BigQuery导入数据，使用Python数值计算库（熊猫，numpy）进行处理，然后将结果写入外部表（Firestore或MySQL ...

数据流由于Window对象或内部处理而丢弃大量事件

最近开发了一个Dataflow使用者，该使用者从PubSub订阅中读取并将组合在同一窗口中的所有这些对象的组合输出到Parquet文件。当我在做...