Обработка сложных сообщений Avro с использованием Kafka Streams - PullRequest
Новая
       56

Обработка сложных сообщений Avro с использованием Kafka Streams

0 голосов
/ 21 января 2019

Я работаю над POC на Kafka Streams, в котором я обрабатываю авро-сообщения с использованием Kafka Streams. Дело в том, что мое сообщение Avro имеет сочетание простых и сложных типов, поэтому мне сложно его обрабатывать.

Моя схема Avro выглядит следующим образом. 

{"type":"record",
"namespace": "com.test",
"name": "backoffice",
"fields": [ {"name": "accountid","type": "string"},
{"name":"amendmentpositionid","type": "int"},
{"name":"booking","type":
{"type":"array","items":
{"namespace":"com.saxo",
"name":"bookingfields",
"type":"record",
"fields":
[{"name":"accountid","type":"string"},{"name":"clientid","type":"int"},
{"name":"clientname","type":"string"},{"name":"exerciseid","type":"int"},
{"name":"hedgeid","type":"int"},{"name":"originatingpositionid","type":"int"},
{"name":"positionid","type":"int"},{"name":"relatedpositionid","type":"int"} ]}}}]}

Входные данные указаны ниже 

{"accountid":"1234","amendmentpositionid":1234,"booking":[{"accountid":"898","clientid":333,"clientname":"Non ","exerciseid":2,"hedgeid":100

Мне нужно, чтобы он выровнялся и выглядел, как указано ниже, прежде чем сохранять его в базе данных. 

1234,1234,898,333,NON,2,100

Чтобы добиться этого, я пытаюсь использовать операцию плоских значений Kafka Streams, но каким-то образом мне не удается сохранить идентификатор и дату в моем конечном выводе.

Мое приложение kafka Streams выглядит ниже. 

package com.test.office.KafkaStreams;

import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializerConfig;
import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde;
import org.apache.avro.generic.GenericData;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
//import io.confluent.kafka;
//import org.apache.kafka.common.serialization.Serdes.
//import io.confluent.kafka.serialiszers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import com.saxo.backoffice;
import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde;
import org.apache.kafka.streams.kstream.KeyValueMapper;

import javax.swing.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class KafkaAvroSchemaRegistry {


    public static void main(String[] args) {

        Properties properties = new Properties();
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "Kafka Avro Topic 8");
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "server1");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
        properties.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://server2:8081");

        KStreamBuilder builder = new KStreamBuilder();
        KStream<String, testSpecific> testspecific1 = builder.stream("topic10");

        KStream<String,testspecific> output1 = testspecific1.peek((key,value) -> System.out.println(key + value.toString()));


output1.print();
KStream<String,String> test = testspecific1.flatMapValues(value -> value.Booking());

test.print()
        KafkaStreams streams = new KafkaStreams(builder, properties);

        streams.cleanUp();
        streams.start();


        // print the topology
       // System.out.println(streams.toString());

        // shutdown hook to correctly close the streams application
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

Может ли кто-нибудь указать мне правильное направление?

1 Ответ

0 голосов
/ 22 января 2019

почему-то я не могу сохранить идентификатор и дату в своем окончательном выводе.

Потому что вы только планируете getMessage().

Попробуйте получить все поля 

flatMapValues(value -> 
    String.format("%d,%s,%s", value.getId(), value.getDate(), value.getMessage());
...