在 Apache Beam 中使用 defaultNaming 进行动态窗口写入答案

【问题标题】：Using defaultNaming for dynamic windowed writes in Apache Beam在 Apache Beam 中使用 defaultNaming 进行动态窗口写入
【发布时间】：2018-10-17 20:01:16
【问题描述】：

我正在关注this post 和the documentation 的答案，以便在管道末端对我的数据执行动态窗口写入。这是我目前所拥有的：

static void applyWindowedWrite(PCollection<String> stream) {
    stream.apply(
        FileIO.<String, String>writeDynamic()
            .by(Event::getKey)
            .via(TextIO.sink())
            .to("gs://some_bucket/events/")
            .withNaming(key -> defaultNaming(key, ".json")));
}

但是 NetBeans 在最后一行警告我语法错误：

FileNaming is not public in Write; cannot be accessed outside package

如何使defaultNaming 可用于我的管道，以便我可以将其用于动态写入。或者，如果这不可能，我应该怎么做？

【问题讨论】：

这看起来不像是光束/数据流问题，而更像是 java 问题。 This 在较早的问题中的回答解释了为什么 java 会抛出这种类型的错误，this one 是一个可能的解决方案。他们中的任何一个有帮助吗？
我同意这是 Java 路径问题。但是，我正在使用 defaultNaming 方法在 Beam 上下文中寻求一些帮助。我参考的帖子显示了这种方法的使用，类似于我使用它的方式，但显然它不会引发相同的错误。我想知道为什么不。

标签： java google-cloud-dataflow apache-beam apache-beam-io

【解决方案1】：

发布我的发现以防其他人遇到此问题。

我之前尝试使用 writeDynamic() 的方式存在三个问题。

之前我一直在使用 Beam 版本 2.3.0，它确实将 FileNaming 描述为 FileIO.Write 的内部类。 Beam 2.4.0 将FileNaming 定义为public static interface，使其在外部可用。
完全解析/导入defaultNaming。而不是直接调用defaultNaming - 如示例文档中所调用的那样 - 它必须作为FileIO.Write.defaultNaming 调用，因为FileIO 是我实际导入的包。
还需要添加withDestinationCoder 才能执行动态写入。

最终的解决方案是这样的。

static void applyWindowedWrite(PCollection<String> stream) {
    stream.apply(FileIO.<String, String>writeDynamic()
                .by(Event::getKey)
                .via(TextIO.sink())
                .to("gs://some_bucket/events/")
                .withDestinationCoder(StringUtf8Coder.of())
                .withNumShards(1)
                .withNaming(key -> FileIO.Write.defaultNaming(key, ".json")));
}

其中Event::getKey 是在同一个包中定义的静态函数，签名为public static String getKey(String event)。

这将执行一个窗口写入，每个窗口写入一个文件（由.withNumShards(1) 方法定义）。这假定窗口已在上一步中定义。在写入之前不需要GroupByKey，因为只要明确定义分片数量，它就会在写入过程中完成。请参阅FileIO documentation 了解“写入文件 -> 每个窗格生成多少分片”下的更多详细信息。

【讨论】：