apache-beam 相关问题

Apache Beam是用于批处理和流处理的统一SDK。它允许使用特定于Beam的DSL指定大规模数据处理工作流程。 Beam工作流可以在不同的运行时执行，如Apache Flink，Apache Spark或Google Cloud Dataflow（云服务）。

错误 DockerEnvironmentFactory：Docker 容器 xxxxx 日志，当尝试使用 Spark 运行器运行用 Go 编写的 Apache Beam 管道时

我有一个用 Go 编写的管道，我想用 Spark 运行器执行，Spark Standalone 安装在我的本地计算机上。阿帕奇光束 2.56.0 阿帕奇火花3.2.2 我启动了 Spark master 并...

docker apache-spark go sdk apache-beam

回答 1 投票 0

带有时基窗口的 Apache 波束流处理过程

我有一个数据流管道，可以从 kafka 读取消息，处理它们，然后将它们插入到 bigquery 中。我希望处理/bigquery 插入将按时间批量进行，这样......

python google-cloud-dataflow apache-beam

回答 1 投票 0

如何为数据流安装Python依赖项

我有一个非常小的python数据流包，包的结构如下所示。 ├── __pycache__ ├── pubsubtobigq.py ├── 需求.txt └── 维尼夫 requirements.txt的内容是原型...

python google-cloud-dataflow apache-beam requirements.txt

回答 1 投票 0

如果我们更新现有管道，Google Dataflow Apache Beam 版本升级将失败

我有一个在 Apache beam Java 2.50.0 上运行的 Google 数据流流管道。我希望通过更新管道选项升级到 2.56.0（当前最新版本）。然而，更新给出了错误......

google-cloud-dataflow apache-beam

回答 1 投票 0

如何在java中配置GCP Spanner ChangeStream读取持续时间

我们在 apache beam java 数据流作业中使用 GCP Spanner Changestream。我们使用 SpannerIO 连接器对其进行配置。代码如下，静态类 Read 扩展 PTransform 我们在 Apache Beam Java 数据流作业中使用 GCP Spanner ChangeStream。我们使用 SpannerIO 连接器对其进行配置。代码如下， static class Read extends PTransform<PBegin, PCollection<DataChangeRecord>> { @Override public PCollection<DataChangeRecord> expand(PBegin input) { Pipeline pipeline = input.getPipeline(); Options options = (Options) pipeline.getOptions(); // Retrieve and parse the startTimestamp and endTimestamp. Timestamp startTimestamp = options.getStartTimestamp().isEmpty() ? Timestamp.now() : Timestamp.parseTimestamp(options.getStartTimestamp()); Timestamp endTimestamp = options.getEndTimestamp().isEmpty() ? Timestamp.now() : Timestamp.parseTimestamp(options.getEndTimestamp()); SpannerConfig spannerConfig = SpannerConfig.create() .withProjectId(getSpannerProjectId(options)) .withInstanceId(getRequiredData(options.getSpannerInstanceId())) .withDatabaseId(getRequiredData(options.getSpannerDatabase())) .withRpcPriority(options.getSpannerRpcPriority()); SpannerIO.ReadChangeStream stream = SpannerIO.readChangeStream() .withSpannerConfig(spannerConfig) .withMetadataInstance(getRequiredData(options.getSpannerMetadataInstanceId())) .withMetadataDatabase(getRequiredData(options.getSpannerMetadataDatabase())) .withChangeStreamName(options.getSpannerChangeStreamName()) .withInclusiveStartAt(startTimestamp) .withInclusiveEndAt(endTimestamp) .withRpcPriority(options.getSpannerRpcPriority()); String spannerMetadataTableName = options.getSpannerMetadataTableName(); if (spannerMetadataTableName != null) { readChangeStream = readChangeStream.withMetadataTable(spannerMetadataTableName); } return pipeline.apply("Read from Spanner", stream ); } 工作正常。但每一秒它都在敲击扳手并寻找变化。我们必须配置扳手点击频率。在日志中我们可以看到启动应用程序后每秒的日志 INFO: Found 0 to be scheduled (readTimestamp = 2024-06-28T23:30:09.117068001Z) Jun 28, 2024 7:30:10 PM org.apache.beam.sdk.io.gcp.spanner.changestreams.action.DetectNewPartitionsAction getAllPartitionsCreatedAfter INFO: Found 0 to be scheduled (readTimestamp = 2024-06-28T23:30:09.117068001Z) Jun 28, 2024 7:30:10 PM org.apache.beam.sdk.io.gcp.spanner.changestreams.action.DetectNewPartitionsAction getAllPartitionsCreatedAfter INFO: Found 0 to be scheduled (readTimestamp = 2024-06-28T23:30:09.117068001Z) Jun 28, 2024 7:30:10 PM org.apache.beam.sdk.io.gcp.spanner.changestreams.action.DetectNewPartitionsAction getAllPartitionsCreatedAfter INFO: Found 0 to be scheduled (readTimestamp = 2024-06-28T23:30:09.117068001Z) 谁知道如何配置这个。我们不想使用 Spanner 数据库客户端。 spannerIO有什么办法吗？提前致谢:) 您看到的日志来自代码中的这一行：https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/ org/apache/beam/sdk/io/gcp/spanner/changestreams/action/DetectNewPartitionsAction.java#L132 此代码与执行变更流查询的Spanner数据库客户端无关。此代码来自 DetectNewPartitionsDoFn，它会轮询元数据表以查找新的更改流分区，以每 100 毫秒执行一次。您能在这里详细说明您的问题吗？

google-cloud-platform apache-beam google-cloud-spanner changestream

回答 1 投票 0

如何处理 Apache Beam (python) 中的异常，以从 JDBC 读取数据并写入 BigQuery

我能够成功从 JDBC 源读取数据，并将输出写回 BigQuery。然而，我仍然坚持寻找处理坏行的 BigQuery 插入异常的最佳方法。例如...

python google-bigquery google-cloud-dataflow apache-beam

回答 1 投票 0

创建经典数据流模板时由于某种原因，模板未写入 template_location

由于某种原因，模板未写入 template_location。当我运行第一个命令时，管道被执行（奇怪），但模板从未生成。可能是什么原因？我是

google-cloud-platform google-cloud-dataflow apache-beam

回答 1 投票 0

beam yaml - 将时间戳添加到 csv 文件

我正在使用beam yaml 创建一个beam 管道。我有一个包含时间戳值的 csv 表，并且想将其用于窗口聚合。不确定使用这个的正确语法是什么