Google Cloud Dataflow是一种完全托管的云服务，可用于大规模创建和评估数据处理管道。数据流管道基于Apache Beam编程模型，可以在批处理和流模式下运行。 Cloud Dataflow是Google云端平台的一部分。
使用 python 运行 gcp 数据流管道时出现 ModuleNotFoundError 消息
我正在尝试在数据流管道中安装依赖项。首先我使用了requirements_file标志，但我得到了（ModuleNotFoundError：没有名为'unidecode'的模块[同时运行'Map（wordcleanfn）-ptransfor...
Python 中 apache Beam 上的 csv 配对出现 UnicodeEncodeError
我正在努力在 apache beam python 中解析 CSV 文件。但是，当 CSV 文件中存在一些 unicode 字符（例如“ş”）时，它无法解析并出现错误 运行时错误：UnicodeEncodeError：'asc...
我使用 Python Apache Beam 创建了自定义数据流模板。在模板逻辑中，我使用 Python 请求调用 Rest api。我想使用 CA 证书对 api 调用进行身份验证，但在数据中...
我正在datalab中“玩”apache beam/dataflow。 我正在尝试从 gcs 读取 csv 文件。 当我使用以下命令创建 pcollection 时： 线= p | 'ReadMyFile' >> beam.io.ReadFromText('gs://' +
在 Apache Beam 和 Dataflow 中使用 ReadFromKafka 时出错
我正在尝试使用 Apache Beam 的 Python SDK 连接到 Kafka 主题，并将管道作为数据流作业提交。 这是我的代码片段 导入系统 导入 apache_beam 作为光束 来自 apache_beam。
如何在使用 apache beam 编写的流式管道中读取 bigquery
我想运行一个从 Google bigquery 表中连续读取的流管道。现在，我的流管道在从 bigquery 表读取一次后停止。 apache beam 文档...
云调度程序在部署数据流作业时出现“INVALID_ARGUMENT”错误（使用 Flex 模板）
我有一个云调度程序，它应该使用 Flex 模板部署数据流作业。我进行了 terraform 更改来创建云调度程序，从而创建云调度程序作业。然而，当我
如何使用 Java 客户端 SDK 列出所有正在运行的 GCP Dataflow 作业
我想知道是否可以使用 Java 客户端 SDK 列出所有正在运行的数据流作业。 我想我可能需要使用： JobsV1Beta3Client jobsV1Beta3Client = JobsV1Beta3Client.create();
apache beam WirteToBigQuery 写入优化
我有一个流管道，我在其中应用 30 秒的窗口并分组在 (“customerId”+“siteId”，element) 元组中。 如何根据公司应用动态表名称...
是否可以将模拟与 doFn 一起使用？我在 doFn 中有一个从 gcs 存储桶读取的处理方法，我想测试一下。 我尝试使用 LocalStorageHelper 但它不支持存储桶获取
如何在传递给 pardo 时修改数据流运行时值提供程序参数？
我在尝试修改传递给 Apache Beam Dataflow 管道中 RuntimeValueProvider 的参数时遇到问题。这是我的代码的简化版本： 导入 apache_beam...
python 数据流：GroupByKey 无法应用于具有全局窗口和默认触发器的无界 PCollection
我有一个简单的 python 数据流代码，它使用无界 pcollection 。它只是 从 pubsub 读取 解析为带有输出标签 SUCCESS 和 FAILURE 的 json 使用输出标签 SUCCESS 和 F 验证 json...
Apache Beam 优化 Firestore 读取 python
我有传感器数据到达 pub/sub (protobuf)，它作为 python 字典插入到“pipeline_fstore”中。数据一次到达。 在管道中，在“添加元数据...
我正在使用 gcloud。 借助 Composer，我拥有了一个可以安排不同任务管道的气流环境。 这些管道之一包括： 为数据流作业做准备的几个任务 数据流...
我有一个写入 BigQuery 表的数据流作业。每个数据流作业都会创建一个新表。 我意识到对 BigQuery 表的写入操作是异步的，即对
我用 Go 编写了一个数据流管道，它从 Pub/Sub 订阅中读取数据，执行转换，然后将记录写入 BigQuery。 它看起来运行良好，但我看不到数据......
在 apache beam ReadFromKafka 中的一个主题中可以确定分区位置之前的超时
我正在从事 Google 数据流工作，我正在使用 apache beam ReadFromKafka 来消费主题消息。我正在消耗 4 个主题。在我们向我们的 kafka clu 添加新的代理之后，管道曾经工作正常......
我正在 Google Dataflow 上使用 Apache Beam，并且正在调用 3 个函数 | "Unnest 1" >> beam.Map(lambda record: dict_level1(record)) | "Unnest 2" >> beam.Map(la...
阿帕奇光束|无法安装 apache beam 并显示子进程错误
我已经安装了最新的 pip、python 3.12，但什么都没有..仍然显示此错误： “为收集的包构建轮子：grpcio-tools grpcio-tools 的构建轮子（setup.py）：已启动
GCP Dataflow 流作业/管道（不成功）耗尽超过 24 小时至 48 小时
我有一个数据流作业，它从 Cloud Spanner 更改流中读取并将其写入 GCS。 我在 36-48 小时前开始耗尽作业，但作业从未离开耗尽状态。 然后我也取消了