在apache beam中通过大型管道选项的最佳实践

Question

我们有一个用例，希望将数百行json规范传递给我们的apache beam管道。？一种直接的方法是创建自定义管道选项，如下所述。有没有其他方法可以将输入作为文件传递？

public interface CustomPipelineOptions extends PipelineOptions {
    @Description("The Json spec")
    String getJsonSpec();
    void setJsonSpec(String jsonSpec);
}

我想在Google数据流引擎中部署管道。即使我将规范作为filepath传递并在启动管道之前读取波束代码中的文件内容，我如何捆绑管道的规范文件部分。注意，我不想在我的波束代码可用的地方提交源代码的规范文件（在资源文件夹中）部分。它需要可配置，即我想为不同的波束管道作业传递不同的规范文件。

Answer 1

Beam管道选项，顾名思义，旨在提供小的配置参数来配置管道<代码>管线选项通常在作业提交时读取。因此，即使您使用管道选项获取json spec to job submission程序，也必须确保您编写的程序在运行时能够访问此文件。为此：

1. 您必须将文件保存在数据流虚拟机可以访问的分布式存储系统中（例如，GCS）

有多种方法（2）。例如

• 直接将文件路径传递给DoFn的构造函数。
• 传入文件路径作为转换的侧输入（允许您在运行时配置它）

Answer 2

您可以将选项作为POJO传递。

public class JsonSpec {
    public String stringArg;
    public int intArg;
}

然后在选项中引用

public interface CustomPipelineOptions extends PipelineOptions {
    @Description("The Json spec")
    JsonSpec getJsonSpec();
    void setJsonSpec(JsonSpec jsonSpec);
}

选项将被解析为类；我相信杰克逊的话，但我不确定。

我想知道为什么要将“数百行JSON”作为管道选项传递？这似乎不是一种很“梁”的做事方式。管道选项应通过配置；您真的需要每个管道运行数百行配置吗？如果您打算传递数据以创建PCollection，那么最好使用TextIO，然后将行处理为JSON。