काफ्का में एवरो सीरियल

यहां मैं एवरो के माध्यम से डेटा को क्रमबद्ध करने और इसे काफ्का में स्थानांतरित करने का एक उदाहरण बताऊंगा। एवरो के लिए काफ्का के लिए एक डेटा धारावाहिक है, यह अपने काम में सर्किट की एक रजिस्ट्री का उपयोग करता है और एक अलग तैनाती सर्वर पर संस्करण का समर्थन करता है। केवल एक धारावाहिक होगा, और यदि आवश्यक हो, तो संस्करण को लागू किया जा सकता है, उदाहरण के लिए, उदाहरण के लिए, डेटाबेस में।

जीथब प्रोजेक्ट

यह एवरो द्वारा तैयार क्रमबद्ध डेटा जैसा लग सकता है। इसमें डेटा और उसके बाद डेटा का हेडिंग विवरण होता है। यह कॉम्पैक्ट और तेज़ निकला, कोई डुप्लिकेट फ़ील्ड नाम नहीं हैं, डेटा प्रारूप द्विआधारी है। डेटा स्कीमा का उपयोग करते समय डेटा की जाँच की जाती है।

उदाहरण सर्किट:

{"namespace": "avro",
"type": "record",
"name": "Person",
"fields": [
     {"name": "name", "type": "string"},
     {"name": "age",  "type": ["int", "null"]}
]
}

स्प्रिंग शेल का उपयोग करते हुए, पहले कमांड में मैं व्यक्तियों की सूची में जोड़ता हूं, एवरो योजना के अनुसार जांच कर रहा हूं:

@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);
    }

GenericRecord एक एवरो रिकॉर्ड है जो योजना के आधार पर बनता है।

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;
    }

}

कंसोल में शेल व्यक्तियों को जोड़ना, और कफका को विषय भेजना:

@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();
    }

यहाँ काफ्का के लिए एवरो धारावाहिक है:

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() {
    }

}

कफका उत्पादक विन्यास:

    @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;
    }

क्रमांकन वर्ग यहाँ निर्दिष्ट किया गया है - "com.example.model.AvroGenericRecordSerializer"
और नया पैरामीटर "SCHEMA" एक स्कीमा ऑब्जेक्ट है, यह द्विआधारी डेटा तैयार करने में AvroGenericRecordSerializer में आवश्यक है।

कंसोल में प्राप्त पक्ष पर, हमें प्राप्त डेटा दिखाई देता है:

एवरो डिसेरिएलाइज़र

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;

    }

}

और कफका उपभोक्ता के समान:

    @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;
    }

Docker wurstmeister / kafka-docker से उपयोग किया जाने वाला काफ्का , आप किसी अन्य का उपयोग कर सकते हैं

जीथब प्रोजेक्ट

avro.apache