JSON用のCreatin kafkaストリームAPI

Question

JSON配列をJSON要素に変換するためのkafkaストリームコードを作成しようとしています...私はkafkaストリームを初めて使用しています。そこkstreamとktable .. と入力の私のストリームにおける次の形式

[ 
{"timestamp":"2017-10-24T12:44:09.359126933+05:30","data":0,"unit":""}, 
{"timestamp":"2017-10-24T12:44:09.359175426+05:30","data":1,"unit":""} 
] 

[ 
{"timestamp":"2017-10-24T12:44:09.359126933+05:30","data":2,"unit":""}, 
{"timestamp":"2017-10-24T12:44:09.359175426+05:30","data":3,"unit":""} 
] 

にできるでしょうし、私の出力は形式でなければなりません

{"timestamp":"2017-10-24T12:44:09.359126933+05:30","data":0,"unit":""} 
{"timestamp":"2017-10-24T12:44:09.359175426+05:30","data":1,"unit":""} 
{"timestamp":"2017-10-24T12:44:09.359126933+05:30","data":2,"unit":""} 
{"timestamp":"2017-10-24T12:44:09.359175426+05:30","data":3,"unit":""} 

は、誰もがコードを書くことで私を助けることができます??

from kafka import KafkaConsumer 
consumer = KafkaConsumer('topicName') 
for message in consumer: 
    print(message) 

はKafkaConsumerでbootstrap_serversパラメータを指定Pythonで

Answer 1

あなたはカフカストリームを使用する場合は、flatMap()を使用することができます。

// using new 1.0 API 
StreamsBuilder builder = new StreamsBuilder(); 
builer.stream("topic").flatMap(...).to("output-topic"); 

のようなものは、より多くの詳細については、例やドキュメントをチェックアウト：

• https://docs.confluent.io/current/streams/developer-guide/index.html
• https://github.com/confluentinc/kafka-streams-examples

Answer 2

...。 Java用

は、本当に良いcloudkarafkaを見て：

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
consumer.subscribe(Arrays.asList(topic)); 
while (true) { 
    ConsumerRecords<String, String> records = consumer.poll(100); 
    for (ConsumerRecord<String, String> record : records) 
     System.out.printf("msg = %s\n", record.value()); 
    } 
}