Как установить AvroCoder с KafkaIO и Apache Beam с Java - PullRequest
Новая
       94

Как установить AvroCoder с KafkaIO и Apache Beam с Java

0 голосов
/ 19 июня 2020

Я пытаюсь создать конвейер, который передает данные из Kafka topi c в BigQuery Google. Данные в топи c есть в Авро.

Я вызываю функцию apply 3 раза. Один раз прочитать из Kafka, один раз извлечь запись и один раз написать в BigQuery. Вот основная часть кода: 

        pipeline
            .apply("Read from Kafka",
                    KafkaIO
                            .<byte[], GenericRecord>read()
                            .withBootstrapServers(options.getKafkaBrokers().get())
                            .withTopics(Utils.getListFromString(options.getKafkaTopics()))
                            .withKeyDeserializer(
                                    ConfluentSchemaRegistryDeserializerProvider.of(
                                            options.getSchemaRegistryUrl().get(),
                                            options.getSubject().get())
                            )
                            .withValueDeserializer(
                                    ConfluentSchemaRegistryDeserializerProvider.of(
                                            options.getSchemaRegistryUrl().get(),
                                            options.getSubject().get()))
                            .withoutMetadata()
            )

            .apply("Extract GenericRecord",
                    MapElements.into(TypeDescriptor.of(GenericRecord.class)).via(KV::getValue)
            )
            .apply(
                    "Write data to BQ",
                    BigQueryIO
                            .<GenericRecord>write()
                            .optimizedWrites()
                            .useBeamSchema()
                            .useAvroLogicalTypes()
                            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                            .withSchemaUpdateOptions(ImmutableSet.of(BigQueryIO.Write.SchemaUpdateOption.ALLOW_FIELD_ADDITION))
                            //Temporary location to save files in GCS before loading to BQ
                            .withCustomGcsTempLocation(options.getGcsTempLocation())
                            .withNumFileShards(options.getNumShards().get())
                            .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
                            .withMethod(FILE_LOADS)
                            .withTriggeringFrequency(Utils.parseDuration(options.getWindowDuration().get()))
                            .to(new TableReference()
                                    .setProjectId(options.getGcpProjectId().get())
                                    .setDatasetId(options.getGcpDatasetId().get())
                                    .setTableId(options.getGcpTableId().get()))

            );

При запуске я получаю следующую ошибку: 

    Exception in thread "main" java.lang.IllegalStateException: Unable to return a default Coder for Extract GenericRecord/Map/ParMultiDo(Anonymous).output [PCollection]. 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: Unable to provide a Coder for org.apache.avro.generic.GenericRecord.
  Building a Coder using a registered CoderProvider failed.

Как настроить кодировщик на правильное чтение Avro?

1 Ответ

0 голосов
/ 21 июня 2020

Есть как минимум три подхода к этому:

  1. Установить кодировщик в строку: 
     pipeline.apply("Read from Kafka", ....)  
    .apply("Dropping key", Values.create())
    .setCoder(AvroCoder.of(Schema schemaOfGenericRecord))
    .apply("Write data to BQ", ....);

Обратите внимание, что ключ отбрасывается, потому что он не используется, с этим вам больше не понадобятся MapElements.

Зарегистрируйте кодировщик в экземпляре конвейера CoderRegistry: 
pipeline.getCoderRegistry().registerCoderForClass(GenericRecord.class, AvroCoder.of(Schema genericSchema));
Получите кодировщик из реестра схем через: 
ConfluentSchemaRegistryDeserializerProvider.getCoder(CoderRegistry registry)

https://beam.apache.org/releases/javadoc/2.22.0/org/apache/beam/sdk/io/kafka/ConfluentSchemaRegistryDeserializerProvider.html#getCoder -org. apache .beam.sdk.coders.CoderRegistry-

...