0

我正在尝试执行 ETL，其中涉及从 HDFS 加载文件、应用转换并将它们写入 Hive。在按照此文档使用 SqlTransforms 执行转换时，我遇到了以下问题。你能帮忙吗？

java.lang.IllegalStateException: Cannot call getSchema when there is no schema
    at org.apache.beam.sdk.values.PCollection.getSchema(PCollection.java:328)
    at org.apache.beam.sdk.extensions.sql.impl.schema.BeamPCollectionTable.<init>(BeamPCollectionTable.java:34)
    at org.apache.beam.sdk.extensions.sql.SqlTransform.toTableMap(SqlTransform.java:105)
    at org.apache.beam.sdk.extensions.sql.SqlTransform.expand(SqlTransform.java:90)
    at org.apache.beam.sdk.extensions.sql.SqlTransform.expand(SqlTransform.java:77)
    at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:537)
    at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:471)
    at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:339)
    at org.apache.beam.examples.SqlTest.runSqlTest(SqlTest.java:107)
    at org.apache.beam.examples.SqlTest.main(SqlTest.java:167)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:282)
    at java.lang.Thread.run(Thread.java:748)

代码片段：

PCollection<String> data = p.apply("ReadLines", TextIO.read().from(options.getInputFile()));

    if(options.getOutput().equals("hive")){
        Schema hiveTableSchema = Schema.builder()
                .addStringField("eid")
                .addStringField("name")
                .addStringField("salary")
                .addStringField("destination")
                .build();
          data.apply(ParDo.of(new DoFn<String, Row>() {
              @ProcessElement
              public void processElement(@Element String input, OutputReceiver<Row> out){
                  String[] values = input.split(",");
                  System.out.println(values);
                  Row row = Row.withSchema(hiveTableSchema)
                                .addValues(values)
                                .build();
                  out.output(row);
              }
          })).apply(SqlTransform.query("select eid, destination from PCOLLECTION"))

                .apply(ParDo.of(new DoFn<Row, HCatRecord>() {
                    @ProcessElement
                    public void processElement(@Element Row input, OutputReceiver<HCatRecord> out){
                        HCatRecord record = new DefaultHCatRecord(input.getFieldCount());
                        for(int i=0; i < input.getFieldCount(); i++){
                            record.set(i, input.getString(i));
                        }
                        out.output(record);
                    }
                        }))
                .apply("WriteData", HCatalogIO.write()
                        .withConfigProperties(configProperties)
                        .withDatabase("wmrpoc")
                        .withTable(options.getOutputTableName()));
4

1 回答 1

0

看起来您需要在PCollection. 在您链接的演练Create...withCoder()中，可以处理该问题。在您的情况下，架构无法从您的 中推断出来ParDo，Beam 可能查看的唯一信息是它输出类型的元素，Row但没有可用的信息，您ParDo甚至是否遵守所有输出的单一架构。所以你需要pcollection.setRowSchema()在申请之前打电话SqlTransform告诉 Beam 你打算从你的转换中得到什么模式ParDo。

更新

并且看起来您之前所做的大部分工作HCatalog最终可能会被简化很多，例如假设您只需要指定类似pipeline.apply(TextIO.readCsvRows(schema)).apply(sqlTransform).... 事实上，Beam SQL 支持读取 CSV 文件而无需额外转换ParDos（通过TextTableProvider），但它尚未连接SqlTransform，只能通过 Beam SQL CLI 访问

于 2018-10-25T16:01:43.703 回答