如何在 Apache Beam 中同时使用 MapElements 和 KV？答案

【问题标题】：How do I use MapElements and KV in together in Apache Beam?如何在 Apache Beam 中同时使用 MapElements 和 KV？
【发布时间】：2018-11-10 02:35:16
【问题描述】：

我想做这样的事情：

PCollection<String> a = whatever;
PCollection<KV<String, User>> b = a.apply(
        MapElements.into(TypeDescriptor.of(KV<String, User>.class))
        .via(s -> KV.of(s, new User(s))));

其中 User 是带有 Arvo 编码器的自定义数据类型和考虑字符串的构造函数。

但是，我收到以下错误：

无法从参数化类型中选择

我尝试将其更改为 TypeDescriptor.of(KV.class)，但随后我得到：

不兼容的类型；必需的 PCollection> 但“应用”被推断为 OutputT：不存在类型变量的实例，因此 PCollection 符合 PCollection>

那么我应该如何将KV 与MapElements 一起使用？

我知道我想要做的是使用ParDo 是可行的，我可以通过清除new DoFn<String, KV<String, User>> 明确指定如何进行类型擦除，但ParDo 不支持lambda 函数。由于我们使用的是 Java 8，这似乎不太优雅....

【问题讨论】：

标签： java java-8 apache-beam

【解决方案1】：

由于在编译期间type erasure in Java，KV<String, User>.class 被转换为KV.class，而在运行时KV.class 没有足够的信息来推断编码器，因为类型变量已被删除。

要绕过这个限制，您需要使用一种在编译后保留类型信息的机制。例如，您可以使用：

TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptor.of(User.class))

这与提供自己的匿名类相同：

new TypeDescriptor<KV<String, User>> {}

提供绑定类型变量的匿名类是目前在 Java 中绕过类型擦除的方法之一。

【讨论】：

【解决方案2】：

尝试使用 SimpleFunction - 它保留类型信息

【讨论】：