Apache - луч Bigquery. От Query ClassCastException

Question

Я пытаюсь выполнить запрос к таблице BigQuery, извлечь один столбец и заполнить файл. Код ниже выдает исключение. Я могу ошибаться, но кажется, что процесс пытается записать временные результаты во временную папку в формате avro, прочитать данные из него и выдает исключение приведения.

pipeLine.apply(
        BigQueryIO.read(
                (SchemaAndRecord elem) -> {
                  GenericRecord record = elem.getRecord();
                  return (String) record.get("column");
                })
                .fromQuery("SELECT column FROM `project.dataset.table`")
                .usingStandardSql()
                .withCoder(AvroCoder.of(String.class)))
        .apply(TextIO.write().to("gs://bucket/test/result/data")
                .withSuffix(TXT_EXT)
                .withCompression(Compression.GZIP));

Причина: java. lang.ClassCastException: org. apache .avro.util.Utf8 нельзя преобразовать в java .lang.String в xxxxx.xxx.xxx.sampling.dataflow.samplingextractor.service.BigQueryExportService.lambda $ export $ 43268ee4 $ 1 ( BigQueryExportService. java: 137) в орг. apache .beam.sdk.io.gcp.bigquery.BigQuerySourceBase $ 1.apply (BigQuerySourceBase. java: 242) в орг. apache .beam.sdk.io .gcp.bigquery.BigQuerySourceBase $ 1.apply (BigQuerySourceBase. java: 235) в орг. apache .beam.sdk.io.AvroSource $ AvroBlock.readNextRecord (AvroSource. java: 597) в орг. apache .beam.sdk.io.BlockBasedSource $ BlockBasedReader.readNextRecord (BlockBasedSource. java: 209) в org. apache .beam.sdk.io.FileBasedSource $ FileBasedReader.advanceImpl (FileBasedSource. java: 484) в орг. apache .beam.sdk.io.FileBasedSource $ FileBasedReader.startImpl (FileBasedSource. * 10 23 *: 479) в орг. apache .beam.sdk.io.OffsetBasedSource $ OffsetBasedReader.start (OffsetBasedSource. java: 249) в орг. apache .beam.runners.dataflow.worker.WorkerCustomSources $ BoundedReaderIterator .start (WorkerCustomSources. java: 601)

Answer 1

При просмотре документации здесь кажется, что вы просто хотите использовать класс StringUtf8Coder .

pipeLine.apply(
    BigQueryIO.read(
            (SchemaAndRecord elem) -> {
              GenericRecord record = elem.getRecord();
              return (String) record.get("column");
            })
            .fromQuery("SELECT column FROM `project.dataset.table`")
            .usingStandardSql()
            .withCoder(StringUtf8Coder.of()))
        .apply(TextIO.write().to("gs://bucket/test/result/data")
            .withSuffix(TXT_EXT)
            .withCompression(Compression.GZIP));

Answer 2

Я думаю, что вы предлагаете использовать .withCoder(AvroCoder.of(org.apache.avro.util.Utf8.class))), поскольку строка не может быть преобразована напрямую из класса Avro Utf8.