如何获取 Apache Beam SQL 查询的输出模式？

Question

我一直在使用 Beam SQL DSL，如果不提供手动了解输出模式的编码器，我将无法使用查询的输出。我可以推断输出模式而不是硬编码吗？

walkthrough 或 examples 均未实际使用查询的输出。我使用 Scio 而不是普通的 Java API 来保持代码相对可读和简洁，我认为这对这个问题没有影响。

这是我的意思的一个例子。

给定一个输入架构 inSchema 和一些映射到 Row 的数据源，如下所示：（在此示例中，基于 Avro，但我认为这并不重要）：

sc.avroFile[Foo](args("input"))
   .map(fooToRow)
   .setCoder(inSchema.getRowCoder)
   .applyTransform(SqlTransform.query("SELECT COUNT(1) FROM PCOLLECTION"))
   .saveAsTextFile(args("output"))

运行此管道会产生KryoException，如下所示：

com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException
Serialization trace:
fieldIndices (org.apache.beam.sdk.schemas.Schema)
schema (org.apache.beam.sdk.values.RowWithStorage)
org.apache.beam.sdk.Pipeline$PipelineExecutionException: 
com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException

但是，插入与 SQL 输出匹配的 RowCoder，在本例中为单个 count int 列：

   ...snip...
   .applyTransform(SqlTransform.query("SELECT COUNT(1) FROM PCOLLECTION"))
   .setCoder(Schema.builder().addInt64Field("count").build().getRowCoder)
   .saveAsTextFile(args("output"))

现在管道运行良好。

鉴于我们指定了输入模式/编码器和查询，因此必须手动告诉管道如何对 SQL 输出进行编码似乎没有必要。在我看来，我们应该能够从中推断出输出模式 - 但除了直接使用 Calcite 之外，我看不出如何？

在 Beam Jira 上提出罚单之前，我想我会检查一下我没有遗漏一些明显的东西！

Answer 1

输出模式推断should work，你的期望是正确的。这似乎是一个错误（在 Beam 或 Scio 中），已提交 BEAM-5335 以供调查。