Google Cloud Dataflow是一种完全托管的云服务，可用于大规模创建和评估数据处理管道。数据流管道基于Apache Beam编程模型，可以在批处理和流模式下运行。 Cloud Dataflow是Google云端平台的一部分。
使用 REST API 创建管道会导致错误 400：INVALID_ARGUMENT
我正在使用 Python 直接调用 Dataflow REST API (https://cloud.google.com/dataflow/docs/reference/data-pipelines/rest/v1/projects.locations.pipelines/create)请求模块来c...
如何处理 Apache Beam (python) 中的异常，以从 JDBC 读取数据并写入 BigQuery
我能够成功从 JDBC 源读取数据，并将输出写回 BigQuery。 然而，我仍然坚持寻找处理坏行的 BigQuery 插入异常的最佳方法。 例如...
创建经典数据流模板时由于某种原因，模板未写入 template_location
由于某种原因，模板未写入 template_location。当我运行第一个命令时，管道被执行（奇怪），但模板从未生成。可能是什么原因？ 我是
是否可以强制数据流作业使用特定版本的Python运行？ 我有一些依赖项仅受 python 3.11 支持
GCP 数据流批处理 MongoDB 到 BigQuery 错误架构与从 TIMESTAMP 更改为 STRING 的字段类型不匹配
我在使用从 Mongo DB 到 BigQuery 的数据流批处理模板时遇到问题。尝试写入 BigQuery 时总是遇到错误。错误是： 提供的架构与表 doj 不匹配...
从 Google 云存储中移动/流式传输数据的最佳方式是什么？另外，Dataflow 是否提供任何功能将数据从 Google 云存储流式传输到外部 GCP？
我在 Azure 中有安全存储帐户，我正在尝试使用 Microsoft Fabric Dataflow Gen2 访问该帐户。我设法使用 vnet 数据网关连接到存储帐户。 获取数据结果...
您好，这可能是愚蠢的问题，但我尚未在任何地方找到答案，目前我需要执行命令来列出数据流作业，如下所示 gcloud 数据流作业列表 --status=done --lim...
Java Apache Beam，使用构造函数变量在 DoFn 的 @Setup 方法中初始化模拟外部客户端
Apache Beam 建议使用 Fakes 而不是 Mocks，因为 Mocks 无法通过管道进行序列化。 我正在为旧代码编写单元测试，其中该类使用 Apache Beam 调用其他扩展...
如何调试在 Google Cloud Dataflow 上未调用 finish_bundle？
这是我在数据流中使用的主要和转换的代码： # main.py 使用 Pipeline(options=options) 作为管道： _ = ( 管道 | “读取输入文件...
缓慢更新侧输入和会话窗口 - 变换节点 AppliedPTransform 未按预期替换
在我的 apache 束流管道中，我有一个与会话窗口一起使用的无限发布/订阅源。 我需要将一些有界配置数据传递到
使用 Apache Beam 进行窗口化 - 修复了窗口似乎没有关闭的问题？
我们正在尝试在 Apache Beam 管道上使用固定窗口（使用 DirectRunner）。我们的流程如下： 从发布/订阅中提取数据 将 JSON 反序列化为 Java 对象 带固定风的窗口事件...
使用 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();