Flink - 如何将 POJO 序列化到 Kafka Sink答案

【问题标题】：Flink - How to serialize a POJO to Kafka SinkFlink - 如何将 POJO 序列化到 Kafka Sink
【发布时间】：2022-02-03 16:38:43
【问题描述】：

我正在尝试 KafkaSink 我的 DataStream，为此我正在使用以下代码：

     KafkaSink<Events> sink = KafkaSink.<Events>builder()
        .setBootstrapServers(LOCAL_KAFKA_BROKER)
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic(OUTPUT_KAFKA_TOPIC)
                .setValueSerializationSchema(new SimpleStringSchema())
                .build())
        .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .build();

在这里，SimpleStringSchema() 不适合，因为我要返回事件的 POJO。这是我一直在使用的 POJO。

    public Events(Date windowStart, Date windowEnd,
                    String metric, String eventId,
                    long count) {
        this.windowStart = windowStart;
        this.windowEnd = windowEnd;
        this.metric = metric;
        this.eventId = eventId;
        this.count = count;
    }

    public Date getWindowStart() {
        return windowStart;
    }

    public void setWindowStart(Date windowStart) {
        this.windowStart = windowStart;
    }

    public Date getWindowEnd() {
        return windowEnd;
    }

    public void setWindowEnd(Date windowEnd) {
        this.windowEnd = windowEnd;
    }

    public String getMetric() {
        return metric;
    }

    public void setMetric(String metric) {
        this.metric = metric;
    }

    public String getEventId() {
        return eventId;
    }

    public void setEventId(String eventId) {
        this.eventId = eventId;
    }

    public long getCount() {
        return count;
    }

    public void setCount(long count) {
        this.count = count;
    }

    public String toString() {
        StringBuilder sb = new StringBuilder("Events{");
        sb.append("windowStart=").append(windowStart);
        sb.append(", windowEnd=").append(windowEnd);
        sb.append(", metric=").append(metric);
        sb.append(", eventId=").append(eventId);
        sb.append(", count=").append(count);
        sb.append("}");

        return sb.toString();
    }

对于 POJO，我无法提出可以在此处使用的 SerializationSchema。我尝试了以下方法：

    public class EventsSerializationSchema implements DeserializationSchema<Events>, SerializationSchema<Events> {

        private static final ObjectMapper objectMapper = new ObjectMapper();
        private String topic;

        public EventsSerializationSchema(){
        }

        public EventsSerializationSchema(String topic) {
            this.topic = topic;
        }

        @Override
        public ProducerRecord<byte[], byte[]> serialize(
                final Events events, @Nullable final Long timestamp) {
            try {
                //if topic is null, default topic will be used
                return new ProducerRecord<>(topic, objectMapper.writeValueAsBytes(events));
            } catch (JsonProcessingException e) {
                throw new IllegalArgumentException("Could not serialize record: " + events, e);
            }
        }
    }

但是，这不起作用，因为我不确定如何序列化它。有人可以帮忙吗？ P.S：由于我使用的是 Flink 1.14，因此 FlinkKafkaPublisher 在此版本中已弃用。

提前致谢

【问题讨论】：

嗯，您是否尝试将new SimpleStringSchema 替换为new EventsSerializationSchema？
否则，如果您先将.map() 一些DataStream<Event> 放入DataStream<String> 然后然后写信给Kafka，SimpleStringSchema 绝对可以正常工作。
是的，我试过了。EventSerializationSchema 不完整，因为它没有序列化方法。这就是我卡住的地方。
你在我的第二条评论中也尝试过解决方案？

标签： java apache-kafka apache-flink

【解决方案1】：

EventsSerializationSchema 实现了错误的接口。您想实现SerializationSchema 或KafkaSerializationSchema，具体取决于您是否愿意实现

byte[] serialize(T element)

或

ProducerRecord<byte[], byte[]> serialize(T element, @Nullable Long timestamp).

有关示例，请参阅 KafkaProducerJob.java 和 UsageRecordSerializationSchema.java。

【讨论】：