{"id":300167,"date":"2020-03-15T21:00:17","date_gmt":"2020-03-15T21:00:17","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=300167"},"modified":"-0001-11-30T00:00:00","modified_gmt":"-0001-11-29T21:00:00","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=300167","title":{"rendered":"Avro serialization \u0432 Kafka"},"content":{"rendered":"\n
\n

\u0417\u0434\u0435\u0441\u044c \u043e\u043f\u0438\u0448\u0443 \u043f\u0440\u0438\u043c\u0435\u0440 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0447\u0435\u0440\u0435\u0437 Avro \u0438 \u043f\u0435\u0440\u0435\u0434\u0430\u0447\u0430 \u0432 Kafka. \u0414\u043b\u044f Avro \u0435\u0441\u0442\u044c \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440 \u0434\u0430\u043d\u043d\u044b\u0445 \u0434\u043b\u044f Kafka, \u043e\u043d \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 \u0432 \u0441\u0432\u043e\u0435\u0439 \u0440\u0430\u0431\u043e\u0442\u0435 \u0440\u0435\u0435\u0441\u0442\u0440 (registry) \u0441\u0445\u0435\u043c \u0438 \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0435\u0442 \u0432\u0435\u0440\u0441\u0438\u043e\u043d\u043d\u043e\u0441\u0442\u044c \u043d\u0430 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u043e\u043c \u0440\u0430\u0437\u0432\u0435\u0440\u043d\u0443\u0442\u043e\u043c \u0441\u0435\u0440\u0432\u0435\u0440\u0435. \u0417\u0434\u0435\u0441\u044c \u0431\u0443\u0434\u0435\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440, \u0430 \u0432\u0435\u0440\u0441\u0438\u043e\u043d\u043d\u043e\u0441\u0442\u044c \u0435\u0441\u043b\u0438 \u043f\u043e\u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f, \u0442\u043e \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440 \u043c\u043e\u0436\u0435\u0442 \u0431\u044b\u0442\u044c \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u0430 \u0441\u0432\u043e\u044f, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440 \u0432 \u0411\u0414. <\/p>\n

<\/a> <\/p>\n

\u041f\u0440\u043e\u0435\u043a\u0442 \u043d\u0430 Github<\/a><\/p>\n

<\/p>\n

\u0412\u043e\u0442 \u0442\u0430\u043a \u043c\u043e\u0436\u0435\u0442 \u0432\u044b\u0433\u043b\u044f\u0434\u0435\u0442\u044c \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043d\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435 \u043f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u043b\u0435\u043d\u043d\u044b\u0435 Avro. \u0415\u0441\u0442\u044c \u0437\u0430\u0433\u043e\u043b\u043e\u0432\u043e\u043a-\u043e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438 \u0434\u0430\u043b\u0435\u0435 \u0441\u0430\u043c\u0438 \u0434\u0430\u043d\u043d\u044b\u0435. \u041f\u043e\u043b\u0443\u0447\u0430\u0435\u0442\u0441\u044f \u043a\u043e\u043c\u043f\u0430\u043a\u0442\u043d\u043e \u0438 \u0431\u044b\u0441\u0442\u0440\u043e, \u043e\u0442\u0441\u0443\u0442\u0441\u0442\u0432\u0443\u044e\u0442 \u043f\u043e\u0432\u0442\u043e\u0440\u044f\u044e\u0449\u0438\u0435\u0441\u044f \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u044f \u043f\u043e\u043b\u0435\u0439, \u0444\u043e\u0440\u043c\u0430\u0442 \u0434\u0430\u043d\u043d\u044b\u0445 \u0431\u0438\u043d\u0430\u0440\u043d\u044b\u0439. \u0414\u0430\u043d\u043d\u044b\u0435 \u043f\u0440\u043e\u0432\u0435\u0440\u044f\u044e\u0442\u0441\u044f \u043f\u0440\u0438 \u0434\u043e\u0431\u0430\u0432\u043b\u0435\u043d\u0438\u0438 \u043f\u043e \u0441\u0445\u0435\u043c\u0435 \u0434\u0430\u043d\u043d\u044b\u0445.<\/p>\n

<\/p>\n

\u041f\u0440\u0438\u043c\u0435\u0440 \u0441\u0445\u0435\u043c\u044b:<\/p>\n

<\/p>\n

{"namespace": "avro", "type": "record", "name": "Person", "fields": [      {"name": "name", "type": "string"},      {"name": "age",  "type": ["int", "null"]} ] }<\/code><\/pre>\n
  <\/p>\n
\u0418\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f Spring Shell, \u0432 \u043f\u0435\u0440\u0432\u043e\u0439 \u043a\u043e\u043c\u0430\u043d\u0434\u0435 \u0434\u043e\u0431\u0430\u0432\u043b\u044f\u044e \u0432 \u0441\u043f\u0438\u0441\u043e\u043a \u043b\u0438\u0446, \u043f\u0440\u043e\u0432\u0435\u0440\u044f\u044f \u043f\u043e \u0441\u0445\u0435\u043c\u0435 Avro:<\/p>\n
  <\/p>\n
@ShellComponent public class Commands {      private List<GenericRecord> records = new ArrayList<>();      @ShellMethod("add user to list for send")     public void add(String name, int age) {         GenericRecord record = new GenericData.Record(SchemaRepository.instance().getSchemaObject());         record.put("name", name);         record.put("age", age);          records.add(record);     }<\/code><\/pre>\n
  <\/p>\n
GenericRecord \u2014 \u044d\u0442\u043e Avro \u0437\u0430\u043f\u0438\u0441\u044c \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u0444\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u0442\u0441\u044f \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0441\u0445\u0435\u043c\u044b<\/p>\n
  <\/p>\n
public class SchemaRepository {      private static final String SCHEMA = "{\\"namespace\\": \\"avro\\",\\n" +             "\\"type\\": \\"record\\",\\n" +             "\\"name\\": \\"Person\\",\\n" +             "\\"fields\\": [\\n" +             "     {\\"name\\": \\"name\\", \\"type\\": \\"string\\"},\\n" +             "     {\\"name\\": \\"age\\",  \\"type\\": [\\"int\\", \\"null\\"]}\\n" +             "]\\n" +             "}\\n";      private static final Schema SCHEMA_OBJECT = new Schema.Parser().parse(SCHEMA);      private static SchemaRepository INSTANCE = new SchemaRepository();      public static SchemaRepository instance() {       return INSTANCE;     }      public Schema getSchemaObject() {         return SCHEMA_OBJECT;     }  }<\/code><\/pre>\n
  <\/p>\n
\u0414\u043e\u0431\u0430\u0432\u043b\u0435\u043d\u0438\u0435 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u0438 shell \u043b\u0438\u0446, \u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0430 \u0432 Kafka \u0442\u043e\u043f\u0438\u043a:<\/p>\n
    <\/p>\n
@ShellComponent public class Commands {      private List<GenericRecord> records = new ArrayList<>();      final private KafkaTemplate template;      public Commands(KafkaTemplate template) {       this.template = template;     }      @ShellMethod("send list users to Kafka")     public void send() {         template.setDefaultTopic("test");         template.sendDefault("1", records);         template.flush();         records.clear();     }<\/code><\/pre>\n
  <\/p>\n
\u0412\u043e\u0442 \u0441\u0430\u043c Avro \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440 \u0434\u043b\u044f Kafka:<\/p>\n
  <\/p>\n
public class AvroGenericRecordSerializer implements Serializer<List<GenericRecord>> {      private Schema schema = null;      @Override public void configure(Map<String, ?> map, boolean b) {         schema = (Schema) map.get("SCHEMA");     }      @Override public byte[] serialize(String arg0, List<GenericRecord> records) {         byte[] retVal = null;          ByteArrayOutputStream outputStream = new ByteArrayOutputStream();         GenericDatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema);          DataFileWriter dataFileWriter = new DataFileWriter<>(datumWriter);         try {             dataFileWriter.create(schema, outputStream);             for (GenericRecord record : records) {                 dataFileWriter.append(record);             }             dataFileWriter.flush();             dataFileWriter.close();             retVal = outputStream.toByteArray();         } catch (IOException e) {             e.printStackTrace();         }         return retVal;     }      @Override public void close() {     }  }<\/code><\/pre>\n
  <\/p>\n
\u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f Kafka producer:<\/p>\n
  <\/p>\n
    @Bean     public Map<String, Object> producerConfigs() {         Map<String, Object> props = new HashMap<>();         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers().get(0));         props.put(ProducerConfig.RETRIES_CONFIG, 0);         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.example.model.AvroGenericRecordSerializer");         props.put("SCHEMA", SchemaRepository.instance().getSchemaObject());         return props;     }<\/code><\/pre>\n
  <\/p>\n
\u0417\u0434\u0435\u0441\u044c \u0443\u043a\u0430\u0437\u0430\u043d \u043a\u043b\u0430\u0441\u0441 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u2014 "com.example.model.AvroGenericRecordSerializer"
  \u0438 \u043d\u043e\u0432\u044b\u0439 \u043f\u0430\u0440\u0430\u043c\u0435\u0442\u0440 "SCHEMA" \u2014 \u044d\u0442\u043e \u043e\u0431\u044a\u0435\u043a\u0442 \u0441\u0445\u0435\u043c\u0430, \u043e\u043d \u043d\u0443\u0436\u0435\u043d \u0432 AvroGenericRecordSerializer \u0432 \u043f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u043a\u0435 \u0431\u0438\u043d\u0430\u0440\u043d\u044b\u0445 \u0434\u0430\u043d\u043d\u044b\u0445<\/p>\n
  <\/p>\n
\u041d\u0430 \u043f\u0440\u0438\u043d\u0438\u043c\u0430\u044e\u0449\u0435\u0439 \u0441\u0442\u043e\u0440\u043e\u043d\u0435 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u0438 \u0432\u0438\u0434\u0438\u043c \u043f\u0440\u0438\u043d\u044f\u0442\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435:<\/p>\n
    <\/p>\n
Avro Deserializer<\/p>\n
  <\/p>\n
public class AvroGenericRecordDeserializer implements Deserializer {      private Schema schema = null;      @Override     public void configure(Map configs, boolean isKey) {         schema = (Schema) configs.get("SCHEMA");     }      @Override     public Object deserialize(String s, byte[] bytes) {         DatumReader<GenericRecord> datumReader = new GenericDatumReader<>(schema);         SeekableByteArrayInput arrayInput = new SeekableByteArrayInput(bytes);         List<GenericRecord> records = new ArrayList<>();          DataFileReader<GenericRecord> dataFileReader = null;         try {             dataFileReader = new DataFileReader<>(arrayInput, datumReader);             while (dataFileReader.hasNext()) {                 GenericRecord record = dataFileReader.next();                 records.add(record);             }         } catch (IOException e) {             e.printStackTrace();         }         return records;      }  }<\/code><\/pre>\n
  <\/p>\n
\u0418 \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u0447\u043d\u044b\u0439 Kafka consumer:<\/p>\n
  <\/p>\n
    @Bean     public Map<String, Object> consumerConfigs() {         Map<String, Object> props = new HashMap<>();         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers().get(0));         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "com.example.model.AvroGenericRecordDeserializer");         props.put("SCHEMA", SchemaRepository.instance().getSchemaObject());         return props;     }<\/code><\/pre>\n
  <\/p>\n
Kafka \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043b \u0438\u0437 Docker wurstmeister\/kafka-docker<\/a>, \u043c\u043e\u0436\u043d\u043e \u043b\u044e\u0431\u0443\u044e \u0434\u0440\u0443\u0433\u0443\u044e<\/p>\n
  <\/p>\n
\u041f\u0440\u043e\u0435\u043a\u0442 \u043d\u0430 Github<\/a><\/p>\n
  <\/p>\n
avro.apache<\/a><\/p>\n<\/div>\n
 \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438  https:\/\/habr.com\/ru\/post\/492312\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"\n
\n
\u0417\u0434\u0435\u0441\u044c \u043e\u043f\u0438\u0448\u0443 \u043f\u0440\u0438\u043c\u0435\u0440 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0447\u0435\u0440\u0435\u0437 Avro \u0438 \u043f\u0435\u0440\u0435\u0434\u0430\u0447\u0430 \u0432 Kafka. \u0414\u043b\u044f Avro \u0435\u0441\u0442\u044c \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440 \u0434\u0430\u043d\u043d\u044b\u0445 \u0434\u043b\u044f Kafka, \u043e\u043d \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 \u0432 \u0441\u0432\u043e\u0435\u0439 \u0440\u0430\u0431\u043e\u0442\u0435 \u0440\u0435\u0435\u0441\u0442\u0440 (registry) \u0441\u0445\u0435\u043c \u0438 \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0435\u0442 \u0432\u0435\u0440\u0441\u0438\u043e\u043d\u043d\u043e\u0441\u0442\u044c \u043d\u0430 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u043e\u043c \u0440\u0430\u0437\u0432\u0435\u0440\u043d\u0443\u0442\u043e\u043c \u0441\u0435\u0440\u0432\u0435\u0440\u0435. \u0417\u0434\u0435\u0441\u044c \u0431\u0443\u0434\u0435\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440, \u0430 \u0432\u0435\u0440\u0441\u0438\u043e\u043d\u043d\u043e\u0441\u0442\u044c \u0435\u0441\u043b\u0438 \u043f\u043e\u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f, \u0442\u043e \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440 \u043c\u043e\u0436\u0435\u0442 \u0431\u044b\u0442\u044c \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u0430 \u0441\u0432\u043e\u044f, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440 \u0432 \u0411\u0414. <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-300167","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/300167","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=300167"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/300167\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=300167"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=300167"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=300167"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}