Kafka flapmapValues 在传递json数组对象时会不会将记录拆分成多条记录？答案

【问题标题】：Will Kafka flapmapValues split the record into multiple records when passing json array object?Kafka flapmapValues 在传递json数组对象时会不会将记录拆分成多条记录？
【发布时间】：2019-01-11 14:15:13
【问题描述】：

我使用的是 confluent 5.0.0 版本*

我有一个如下所示的 JSON 数组：

{ 
    "name" : "David,Corral,Babu", 
    "age" : 23
}

并且通过使用 kafka 流，我想根据“name”键值中的逗号标准将上述记录分成两个。输出应该是这样的：

{ 
    "name" : "David", 
    "age" : 23
},
{ 
    "name" : "Corral", 
    "age" : 23
},
{
    "name" : "Babu", 
    "age" : 23
 }

为此，我使用“flatMapValues”。但到目前为止我无法实现预期的结果。

但想检查“flatmapValues”是否是要使用的正确函数满足我的要求？

我使用了以下代码：

package test;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.Produced;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public class RecordSplliter {

    public static void main(String[] args) throws Exception {
        System.out.println("** STARTING RecordSplliter STREAM APP **");
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "json-e44nric2315her");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, PersonSeder.class);

        final Serde<String> stringSerde = Serdes.String();
        final StreamsBuilder builder = new StreamsBuilder();

        // Consume JSON and enriches it
        KStream<String, Person> source = builder.stream("streams-plaintext-input");

        KStream<String, String> output = source
            .flatMapValues(person -> Arrays.asList(person.getName().split(",")));
        output.to("streams-output");

        final Topology topology = builder.build();
        final KafkaStreams streams = new KafkaStreams(topology, props);
        final CountDownLatch latch = new CountDownLatch(1);

        // Attach shutdown handler to catch control-c
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });

        try {
            streams.start();
            latch.await();
        } catch (Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
}

在运行时出现以下异常：

    08:31:10,822 ERROR 
    org.apache.kafka.streams.processor.internals.AssignedStreamsTasks  - 
    stream-thread [json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387- 
    StreamThread-1] Failed to process stream task 0_0 due to the following 
    error:
    org.apache.kafka.streams.errors.StreamsException: Exception caught in 
    process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=streams- 
    plaintext-input, partition=0, offset=0
    at 
 org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:304)
    at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:409)
    at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:957)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:832)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: myapps.PersonSerializer) is not compatible to the actual key or value type (key type: unknown because key is null / value type: java.lang.String). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
    at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:94)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.kstream.internals.KStreamFlatMapValues$KStreamFlatMapValuesProcessor.process(KStreamFlatMapValues.java:42)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:288)
    ... 6 more
Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to myapps.Person
    at myapps.PersonSerializer.serialize(PersonSerializer.java:1)
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:154)
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:98)
    at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:89)
    ... 18 more
08:31:10,827 INFO  org.apache.kafka.streams.processor.internals.StreamThread     - stream-thread [json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN
08:31:10,827 INFO  org.apache.kafka.streams.processor.internals.StreamThread     - stream-thread [json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387-StreamThread-1] Shutting down
08:31:10,833 INFO  org.apache.kafka.clients.producer.KafkaProducer               - [Producer clientId=json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387-StreamThread-1-producer] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
08:31:10,843 INFO  org.apache.kafka.streams.processor.internals.StreamThread     - stream-thread [json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD
08:31:10,843 INFO  org.apache.kafka.streams.KafkaStreams                         - stream-client [json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387] State transition from RUNNING to ERROR
08:31:10,843 WARN  org.apache.kafka.streams.KafkaStreams                         - stream-client [json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387] All stream threads have died. The instance will be in error state and should be closed.
08:31:10,843 INFO  org.apache.kafka.streams.processor.internals.StreamThread     - stream-thread [json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387-StreamThread-1] Shutdown complete
Exception in thread "json-enricher-0f8bc964-40c0-41f2-a724-dfa638923387-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=streams-plaintext-input, partition=0, offset=0
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:304)
    at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:409)
    at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:957)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:832)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: myapps.PersonSerializer) is not compatible to the actual key or value type (key type: unknown because key is null / value type: java.lang.String). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
    at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:94)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.kstream.internals.KStreamFlatMapValues$KStreamFlatMapValuesProcessor.process(KStreamFlatMapValues.java:42)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:288)
    ... 6 more
Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to myapps.Person
    at myapps.PersonSerializer.serialize(PersonSerializer.java:1)
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:154)
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:98)
    at 
 org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:89)
    ... 18 more

【问题讨论】：

标签： java json apache-kafka apache-kafka-streams

【解决方案1】：

例外是因为您的flatMapValues 产生了String 类型的值。在您的代码中，您没有将任何Produced 传递给KStream::to 函数，因此它会尝试使用默认值（在属性中传递），在您的情况下为PersonSeder.class。

您的值是String 类型，但PersonSeder.class 用于序列化。

如果你想拆分它，你需要这样的东西

KStream<String, Person> output = source
    .flatMapValues(person -> 
        Arrays.stream(person.getName().split(","))
            .map(name -> new Person(name, person.getAge()))
            .collect(Collectors.toList()));

我在您的序列化器和反序列化器中使用了以下代码，它是对称的（也使用 Gson）并且可以正常工作

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app1");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, PersonSerdes.class);
    final StreamsBuilder builder = new StreamsBuilder();
    KStream<String, Person> source = builder.stream("input");
    KStream<String, Person> output = source
            .flatMapValues(person ->
                    Arrays.stream(person.getName()
                            .split(","))
                            .map(name -> new Person(name, person.getAge()))
                            .collect(Collectors.toList()));
    output.to("output");

    KafkaStreams streams = new KafkaStreams(builder.build(), props);
    streams.start();
    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

更新 1：

根据您关于使用 json 而不是 POJO 的问题，一切都取决于您的 Sedes。如果你使用 Generic Serdes，你可以序列化和反序列化到 Json (Map)

以下是简单的 MapSerdes，可用于此目的和示例代码。

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;

import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.Map;

public class MapSerdes implements Serde<Map<String, String>> {

    private static final Charset CHARSET = Charset.forName("UTF-8");

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {}

    @Override
    public void close() {}

    @Override
    public Serializer<Map<String, String>> serializer() {
        return new Serializer<Map<String, String>>() {
            private Gson gson = new Gson();
            @Override
            public void configure(Map<String, ?> configs, boolean isKey) {}

            @Override
            public byte[] serialize(String topic, Map<String, String> data) {
                String line = gson.toJson(data); // Return the bytes from the String 'line'
                return line.getBytes(CHARSET);
            }

            @Override
            public void close() {}
        };
    }

    @Override
    public Deserializer<Map<String, String>> deserializer() {
        return new Deserializer<Map<String, String>>() {
            private Type type = new TypeToken<Map<String, String>>(){}.getType();
            private Gson gson = new Gson();
            @Override
            public void configure(Map<String, ?> configs, boolean isKey) {}

            @Override
            public Map<String, String> deserialize(String topic, byte[] data) {
                Map<String,String> result = gson.fromJson(new String(data), type);
                return result;
            }

            @Override
            public void close() {}
        };
    }
}

示例用法：取而代之的是名称，取决于您的地图，您可以使用不同的属性。

public class GenericJsonSplitterApp {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app1");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, MapSerdes.class);

        final StreamsBuilder builder = new StreamsBuilder();
        KStream<String, Map<String, String>> source = builder.stream("input");
        KStream<String, Map<String, String>> output = source
                .flatMapValues(map ->
                        Arrays.stream(map.get("name")
                                .split(","))
                                .map(name -> {
                                    HashMap<String, String> splittedMap = new HashMap<>(map);
                                    splittedMap.put("name", name);
                                    return splittedMap;
                                })
                                .collect(Collectors.toList()));
        output.to("output");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

【讨论】：

仍然出现同样的错误：原因：org.apache.kafka.streams.errors.StreamsException：序列化程序（键：org.apache.kafka.common.serialization.StringSerializer / 值：myapps. PersonSerializer）与实际的键或值类型不兼容（键类型：未知，因为键为空/值类型：java.lang.String）。更改 StreamConfig 中的默认 Serdes 或通过方法参数提供正确的 Serdes。原因：java.lang.ClassCastException：java.lang.String 无法在 myapps.PersonSerializer.serialize(PersonSerializer.java:1) 处强制转换为 myapps.Person
代码如下：-props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, PersonSeder.class); final StreamsBuilder builder = new StreamsBuilder(); KStream<String, Person> source = builder.stream("streams-plaintext-input"); KStream<String, Person> output = source.flatMapValues(person -> Arrays.stream(person.getName().split(",")) .map(name -> new Person(name, person.getAge())) .collect(Collectors.toList())); output.to("streams-output");
@Jana，您能否添加您的序列化程序的代码 - PersonSerializer，我认为可能有问题？
public class PersonSerializer implements Closeable, AutoCloseable, Serializer<Person> { private static final Charset CHARSET = Charset.forName("UTF-8"); static private Gson gson = new Gson(); @Override public void configure(Map<String, ?> map, boolean b) { } @Override public byte[] serialize(String s, Person person) { // Transform the Person object to String String line = gson.toJson(person); // Return the bytes from the String 'line' return line.getBytes(CHARSET); } @Override public void close() { }}
@Jana，我已经添加了我使用的整个代码，你能和你的比较一下吗？