Google Cloud Dataflow是一种完全托管的云服务，可用于大规模创建和评估数据处理管道。数据流管道基于Apache Beam编程模型，可以在批处理和流模式下运行。 Cloud Dataflow是Google云端平台的一部分。
我在 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();
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，我拥有了一个可以安排不同任务管道的气流环境。 这些管道之一包括： 为数据流作业做准备的几个任务 数据流...