Hive 3.1+ не десериализует сообщения Avro 1.8.3+ от Kafka 1.0+ - PullRequest
Новая
       101

Hive 3.1+ не десериализует сообщения Avro 1.8.3+ от Kafka 1.0+

0 голосов
/ 05 февраля 2019

Допустим, у меня есть тема, созданная с помощью потоков kafka из Confluent, которая содержит сообщения, сериализованные в avro с io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer

Затем я создаю внешнюю таблицу kafka в Hive 

CREATE EXTERNAL TABLE k_table
(`id` string , `sequence` int) 
STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler' 
TBLPROPERTIES 
( 
  "kafka.topic" = "sample-topic", 
  "kafka.bootstrap.servers"="kafka1:9092", 
  "kafka.serde.class"="org.apache.hadoop.hive.serde2.avro.AvroSerDe", 
  "avro.schema.url"="Sample.avsc"

);

Когда я запускаю запрос: 

select * from k_table WHERE `__timestamp` >  1000 * to_unix_timestamp(CURRENT_TIMESTAMP - interval '2' DAYS)

Я получил неожиданную ошибку ввода-вывода: 

INFO  : Executing command(queryId=root_20190205160129_4579b5ff-9a5c-496d-8d03-9a7ccc0f6d90): select * from k_tickets_prod2 WHERE `__timestamp` >  1000 * to_unix_timestamp(CURRENT_TIMESTAMP - interval '1' minute)
INFO  : Completed executing command(queryId=root_20190205160129_4579b5ff-9a5c-496d-8d03-9a7ccc0f6d90); Time taken: 0.002 seconds
INFO  : Concurrency mode is disabled, not creating a lock manager
Error: java.io.IOException: java.lang.ArrayIndexOutOfBoundsException: 55 (state=,code=0)

Хорошо все работает нормально с потребителем Confluent kafka, а также я попытался установитьДесериализатор сливного кафки в TBLPROPERTIES, который, похоже, должен действовать.

Среда: 

Hive 4.0 + Beeline 3.1.1 + Kafka 1.1 (Clients & Broker) + Confluent 4.1

1 Ответ

0 голосов
/ 05 февраля 2019

Проблема в том, что производитель Confluent сериализует авро-сообщения в произвольном формате как <magic_byte 0x00><4 bytes of schema ID><regular avro bytes for object that conforms to schema>.Таким образом, у обработчика Hive kafka есть проблема с десериализацией, потому что он использует базовый десериализатор kafka bytearray, и эти 5 байтов в начале сообщения являются неожиданными.

Я создал ошибку в кусте для поддержки формата Confluent и реестра Schema, а также сделал PR с быстрым исправлением , удаляющим 5 байт.из сообщения после "avro.serde.magic.bytes"="true" свойство установлено в TBLPROPERTIES.

После этого патча оно работает как шарм.

...