标签: apache-beam

我想从在谷歌数据流上运行的 apache beam 管道连接谷歌云 sql postgres 实例。
我想使用 Python SDK 来完成此操作。
我无法为此找到适当的文档。
在云SQL如何指导我没有看到任何数据流文档。
https://cloud.google.com/sql/docs/postgres/

有人可以提供文档链接/github 示例吗？

标题几乎概括了一切。无论我将窗口宽度设置得有多短，GroupByKey在 DirectRunner 上运行作业时都不会触发。使用 DataflowRunner 时，一切都按预期工作。

Apache Beam 2.1.0 存在从 BigQuery 读取的模板管道的错误，这意味着它们只能执行一次。更多详细信息请参见https://issues.apache.org/jira/browse/BEAM-2058

Beam 2.2.0 的发布已修复此问题，您现在可以使用withTemplateCompatibility选项从 BigQuery 读取数据，您的模板管道现在可以多次运行。

  pipeline
    .apply("Read rows from table."
         , BigQueryIO.readTableRows()
                     .withTemplateCompatibility()
                     .from("<your-table>")
                     .withoutValidation())

这种实现似乎给 BigQueryIO 读取操作带来了巨大的性能成本，我现在的批处理管道在8-11 分钟内运行，现在始终需要45-50 分钟才能完成。两个管道之间的唯一区别是.withTemplateCompatibility()。

我试图了解性能大幅下降的原因以及是否有任何方法可以改进它们。

谢谢。

解决方案：基于jkff的输入。

  pipeline
    .apply("Read rows from table."
         , BigQueryIO.readTableRows()
                     .withTemplateCompatibility()
                     .from("<your-table>")
                     .withoutValidation())
    .apply("Reshuffle",  Reshuffle.viaRandomKey())

我有一个线性的三步数据流管道——出于某种原因，最后一步开始了，但Not started在我放弃并终止这项工作之前，前两步挂了很长时间。我不确定是什么原因造成的，因为这个相同的管道在过去已经成功运行，而且我很惊讶它没有在日志中显示任何关于阻止前两个步骤开始的错误。什么会导致这种情况，我该如何防止它发生？

使用Direct Runner在本地运行 Apache Beam 管道时，日志级别似乎设置为DEBUG.

有没有办法设置日志来INFO代替？

注意：--workerLogLevelOverrides可以在使用 Cloud Dataflow Runner 时使用，但似乎不适用于 Direct Runner

使用 Apache Beam Python SDK 版本 2.9.0，是否可以获得类似于 Google\xe2\x80\x99s 数据流的可渲染管道图表示，而不是运行它？

我很难组装复杂的管道，并且在尝试使用 执行它之前我很高兴看到组装好的管道DirectRunner。

我的目标是从云存储中读取 avro 文件数据并使用 Java 将其写入 BigQuery 表。如果有人提供代码片段/想法来读取 avro 格式数据并使用 Cloud Dataflow 将其写入 BigQuery 表，那就太好了。

如何实现位于https://beam.apache.org/documentation/pipelines/design-your-pipeline/的以下逻辑：

//merge the two PCollections with Flatten//me 
PCollectionList<String> collectionList = PCollectionList.of(aCollection).and(bCollection);
PCollection<String> mergedCollectionWithFlatten = collectionList
    .apply(Flatten.<String>pCollections());

// continue with the new merged PCollection
mergedCollectionWithFlatten.apply(...);

从而可以将多个 PCollection 组合成 apache beam python api 中的单个 PCollection？

我正在阅读一个逗号分隔的 CSV 文件，其中的字段用双引号括起来，其中一些字段的值中也有逗号，例如： "abc","def,ghi","jkl"

有什么方法可以使用 Apache Beam 将此文件读入 PCollection 吗？

我将如何使用 Spring 执行到 Google Cloud Dataflow 的 Apache Beam 管道？这个问题类似于在 Google Data Flow 上的 Spring Boot 项目中运行 Apache Beam 管道，但这个问题更关注从 Spring 控制器而不是从 CommandLineRunner 启动管道。

@RestController
@RequestMapping("/task/import-csv-file")
public class ImportCsvController {
    @PostMapping("/process-csv-file")
    private ResponseEntity<Void> processCsvFile(
            @RequestParam String gcsFileName,
            @RequestParam String bucketName
    ) {
        DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);

        options.setProject("same-project-as-this-app-engine-instance");
        options.setStagingLocation("gs://" + bucketName + "/binaries");
        options.setRunner(DataflowRunner.class);
        options.setJobname("process-csv");

        Pipeline pipeline = Pipeline.create(options);
        pipeline.apply("ReadFile", TextIO.read().from("gs://" + bucketName + "/" + gcsFileName));
        // ... apply some more transforms here, which will eventually 
        // write …