我有一个TableRow
的PCollection
。我想在那个PCollection
上使用SQLTransformation
。我试图在PCollection<TableRow>
上直接使用SQLTransformation
,但是它给出了下面的错误:
- 代码段**
PCollection<TableRow> rows = [...]
PCollection<Row> rows1 = rows.apply(SqlTransform.query("SELECT max(ID) as max_watermark FROM PCOLLECTION"));
- 错误**
[main] WARN org.apache.beam.sdk.io.jdbc.JdbcIO - Unable to infer a schema for type com.google.api.services.bigquery.model.TableRow. Attempting to infer a coder without a schema.
java.lang.IllegalStateException: Cannot call getSchema when there is no schema
at org.apache.beam.sdk.values.PCollection.getSchema(PCollection.java:331)
at org.apache.beam.sdk.extensions.sql.impl.schema.BeamPCollectionTable.<init>(BeamPCollectionTable.java:35)
at org.apache.beam.sdk.extensions.sql.SqlTransform.toTableMap(SqlTransform.java:183)
at org.apache.beam.sdk.extensions.sql.SqlTransform.expand(SqlTransform.java:138)
at org.apache.beam.sdk.extensions.sql.SqlTransform.expand(SqlTransform.java:110)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:548)
at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:482)
at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:360)
at org.example.Main.main(Main.java:162)
根据文档,SQLTransformation需要PCollection<Row>
作为输入,因此我尝试使用以下逻辑将PCollection<TableRow>
转换为PCollection<Row>
,但这会导致编码器错误。
"我试过逻辑"
final Schema schema = Schema.builder()
.addStringField("ID")
.build();
PCollection<Row> rows11 = rows.apply(ParDo.of(new DoFn<TableRow, Row>() {
@ProcessElement
public void ProcessElement(@Element TableRow inRow, OutputReceiver<Row> out){
Row r = Row.withSchema(schema)
.addValues(inRow.get("ID"))
.build();
out.output(r);
}
}));
PCollection<Row> rows12 = rows11.apply(SqlTransform.query("SELECT max(ID) as max_watermark FROM PCOLLECTION"));
但这会产生下面的错误。我不知道我在这里遗漏了什么。* 我完整的用例是从rows(PCollection<TableRow>
)获取id列值的最大值,并将其存储在BigQuery表之一中。*
- 错误**
[main] WARN org.apache.beam.sdk.io.jdbc.JdbcIO - Unable to infer a schema for type com.google.api.services.bigquery.model.TableRow. Attempting to infer a coder without a schema.
java.lang.IllegalStateException: Unable to return a default Coder for ParDo(Anonymous)/ParMultiDo(Anonymous).output [PCollection@185939155]. Correct one of the following root causes:
No Coder has been manually specified; you may do so using .setCoder().
Inferring a Coder from the CoderRegistry failed: Cannot provide a coder for a Beam Row. Please provide a schema instead using PCollection.setRowSchema.
Using the default output Coder from the producing PTransform failed: PTransform.getOutputCoder called.
at org.apache.beam.sdk.util.Preconditions.checkStateNotNull(Preconditions.java:471)
at org.apache.beam.sdk.values.PCollection.getCoder(PCollection.java:284)
at org.apache.beam.sdk.values.PCollection.finishSpecifying(PCollection.java:115)
at org.apache.beam.sdk.runners.TransformHierarchy.finishSpecifyingInput(TransformHierarchy.java:154)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:547)
at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:482)
at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:360)
at org.example.Main.main(Main.java:159)
1条答案
按热度按时间jjjwad0x1#
结果
PCollection<Row>
中缺少.setCoder(RowCoder.of(schema))
。它们也是在PCollection级别定义的。有关https://beam.apache.org/documentation/programming-guide/#schemas详细信息,请访问www.example.com。