Kafka Avro Consumer with Decoder issues

Question

Khi tôi cố gắng chạy Kafka Consumer with Avro qua dữ liệu với lược đồ tương ứng, nó trả về lỗi "AvroRuntimeException: Dữ liệu bị sai. Độ dài là âm: -40". Tôi thấy những người khác gặp sự cố tương tự coverting byte array to json, Avro write and read và Kafka Avro Binary *coder. Tôi cũng đã tham chiếu đến số Consumer Group Example này, tất cả đều hữu ích, tuy nhiên không có trợ giúp nào về lỗi này cho đến nay .. Nó hoạt động cho đến khi mã này (dòng 73)

Decoder decoder = DecoderFactory.get(). BinaryDecoder (byteArrayInputStream, null);

Tôi đã thử các bộ giải mã khác và in ra nội dung của biến byteArrayInputStream trông như thế nào tôi tin bạn sẽ mong đợi dữ liệu avro được tuần tự hóa (trong thông báo tôi có thể thấy lược đồ và một số dữ liệu và một số dữ liệu không đúng định dạng). in ra các Bytes có sẵn bằng cách sử dụng phương thức .available(), trả về 594. Tôi không hiểu tại sao lỗi này lại xảy ra. Apache Nifi được sử dụng để tạo luồng Kafka với cùng một lược đồ từ hdfs. Tôi sẽ đánh giá cao sự giúp đỡ nào.

Answer 1

Có lẽ sự cố không khớp giữa cách dữ liệu Avro được viết (được mã hóa) bởi Nifi so với cách ứng dụng người tiêu dùng đọc (giải mã) dữ liệu.

Tóm lại, API Avro của cung cấp hai cách tiếp cận khác nhau để serialization:

1. Để tạo thích hợp Avro file: Để mã hóa các bản ghi dữ liệu mà còn để nhúng các schema Avro trong một loại lời mở đầu (thông qua org.apache.avro.file.{DataFileWriter/DataFileReader}). Việc nhúng lược đồ vào các tệp Avro có ý nghĩa rất nhiều vì (a) thường là "tải trọng" của các tệp Avro là các đơn vị lớn hơn lược đồ Avro được nhúng và (b) bạn có thể sao chép hoặc di chuyển các tệp đó trong nội dung trái tim của bạn và vẫn chắc chắn rằng bạn có thể đọc lại chúng mà không phải hỏi ý kiến ​​ai đó hoặc điều gì đó.
2. Để chỉ mã hóa các bản ghi dữ liệu, tức là không nhúng lược đồ (qua org.apache.avro.io.{BinaryEncoder/BinaryDecoder}; hãy lưu ý sự khác biệt trong tên gói: io tại đây so với file ở trên). Cách tiếp cận này thường được ưu tiên khi các thông điệp mã hóa Avro được viết cho chủ đề Kafka, ví dụ, bởi vì so với biến thể 1 ở trên, bạn không phải chịu chi phí cho việc nhúng lại lược đồ Avro vào mỗi thư, giả sử rằng (rất hợp lý) chính sách là, đối với cùng một chủ đề Kafka, các tin nhắn được định dạng/mã hóa với cùng một lược đồ Avro. Đây là một lợi thế đáng kể bởi vì trong ngữ cảnh dữ liệu luồng, dữ liệu dữ liệu chuyển động thường nhỏ hơn nhiều (thường là từ 100 byte đến vài trăm KB) so với các tệp Avro dữ liệu còn lại như được mô tả ở trên (thường là hàng trăm hoặc hàng nghìn MB); do đó kích thước của lược đồ Avro là tương đối lớn, và do đó bạn không muốn nhúng nó 2000x khi ghi 2000 bản ghi dữ liệu vào Kafka. Nhược điểm là bạn phải "theo cách nào đó" theo dõi cách các lược đồ Avro ánh xạ tới chủ đề Kafka - hay chính xác hơn, bằng cách nào đó bạn phải theo dõi lược đồ Avro nào được mã hóa mà không đi xuống đường dẫn nhúng lược đồ trực tiếp. Tin vui là có tooling available in the Kafka ecosystem (Avro schema registry) để làm điều này một cách minh bạch. Vì vậy, so với biến thể 1, biến thể 2 tăng hiệu quả với chi phí tiện lợi.

Hiệu ứng là "định dạng dây" cho dữ liệu Avro được mã hóa sẽ khác nhau tùy thuộc vào việc bạn sử dụng (1) hoặc (2) ở trên.

Tôi không quen với Apache Nifi, nhưng xem nhanh mã nguồn (ví dụ: ConvertAvroToJSON.java) gợi ý rằng nó đang sử dụng biến thể 1, tức là nó nhúng lược đồ Avro cùng với bản ghi Avro. Mã người tiêu dùng của bạn, tuy nhiên, sử dụng DecoderFactory.get().binaryDecoder() và do đó biến thể 2 (không có lược đồ được nhúng).

Có lẽ điều này giải thích lỗi bạn đang gặp phải?