Чтение файлов AVRO в GCS как PCOLLECTION ⇐ JAVA

[Anonymous]

Наша задание Dataflow, написанная на Python, слушает из подписки PubSub. Сообщения представляют собой строки пути файла GCS файлов AVRO (

Код: Выделить всё

gs://bucket/file-timestamp.avro

). Файлы AVRO не имеют единой схемы, но Avroio.ReachallFromavro Beam Python достаточно хорош, чтобы в целом анализировать записи файла AVRO в GCS в качестве словарей.

Код: Выделить всё

# Python code
pipeline
| "Read from Pub/Sub" >> beam.io.ReadFromPubSub(subscription=input_subscription)
| "Read Avro Files" >> avroio.ReadAllFromAvro(with_filename=True)

Ожидаемая форма полученной Ptransform - это кортеж: (some_gcs_filepath, record_as_dict)
Из -за некоторых ограничений в Python мы должны преобразовать наш код в Java. Мы можем прочитать из Pubsub Light, однако, мы еще не знаем, как использовать Avroio , чтобы получить результирующую PCOLLECTION без указания схемы.

Код: Выделить всё

// Java code
pipeline
.apply("Read from Pub/Sub", PubsubIO.fromSubscription(options.getInputSubscription()))
.apply("Read Avro Files", AvroIO.__________________)

Было бы лучше, если бы исходный gcs_filepath передаст вместе с ним так же, как и то, что мы делаем в Python. Нужно ли использовать Fileio в тандеме с Avroio ? Как мне это достичь?

Код: Выделить всё

pipeline
.apply("Read from Pub/Sub", PubsubIO.fromSubscription(options.getInputSubscription()))
.apply("Read Avro Files", FileIO.matchAll())
.apply(FileIO.readMatches())
.apply(AvroIO.parseFilesGenericRecords(record -> record))
< /code>
Код выше приведет меня к некоторому прогрессу, но тогда жалуется, что я должен предоставить кодировщик < /p>
java.lang.IllegalArgumentException: Unable to infer coder for output of parseFn. Specify it explicitly using withCoder()
< /code>
, который в последней строке должен выглядеть следующим образом: < /p>
  .apply(AvroIO.parseFilesGenericRecords(record -> record).withCoder(someCoder))

Но тогда Somecoder требует конкретной схемы, которую я не могу предоставить, поскольку входящий файл AVRO может быть разной схемой.

Подробнее здесь: https://stackoverflow.com/questions/747 ... ericrecord