flink1.12.x kafka反序列化器JSONKeyValueDeserializationSchema

作者：马育民 • • 阅读：10239

说明

flink作为消费者，读取kafka消息时，需要指定 反序列化器

一般 kafka中的数据都是 json格式，flink内置了 json反序列化器：JSONKeyValueDeserializationSchema

传统方式

大多数的程序员都会使用 new SimpleStringSchema() 作为 反序列化器，读取Kafka中的数据，然后使用 alibaba 提供的 fastJson 来解析数据（或者使用 jackson 解析数据）

缺点：

  • 接受的json数据字段不稳定，有的数据多一个字段，有的业务少一个字段
  • 有的字段值为null，可能出现空指针异常

JSONKeyValueDeserializationSchema 优点

  • 解析json结构
  • 避免出现空指针异常
  • 可以读取 kafka 元数据

读取的格式如下

{
    "value":{
        "name":"李雷",
        "url":"/jianpan.html",
        "time":"2020-09-01 10:00:00"
    },
    "metadata":{ // 元数据
        "offset":643, // 偏移量
        "topic":"kafka2flink", // 主题
        "partition":0 // 分区
    }
}

api

创建 JSONKeyValueDeserializationSchema

public JSONKeyValueDeserializationSchema(boolean includeMetadata)

形参：

  • includeMetadata：boolean类型，
    • 为true表示需要指明是否需要包含“元数据”、偏移量、分区和主题等信息
    • 为false表明只需要数据

ObjectNode

将 json格式的字符串，反序列化为 ObjectNode 对象

可以用 objectNode.get("field") 访问字段。

案例

package source;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

public class TestFilter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<ObjectNode> fkc = new FlinkKafkaConsumer("kafka2flink", new JSONKeyValueDeserializationSchema(true), pro);
        DataStreamSource<ObjectNode> dss = env.addSource(fkc);

        SingleOutputStreamOperator<UserLog> mapSOSO = dss.map(new MyMap());

        mapSOSO.print();

        env.execute();
    }

    /**
     * 第一个泛型，表示原数据的类型
     * 第二个泛型，表示转换后的数据类型
     */
    public static class MyMap implements MapFunction<ObjectNode, UserLog>{

        @Override
        public UserLog map(ObjectNode data) throws Exception {

            JsonNode name = data.get("value").get("name");
            JsonNode url = data.get("value").get("url");
            JsonNode time = data.get("value").get("time");

            UserLog result = new UserLog();
            if(name!=null) {
                result.setName(name.asText());
            }
            if(url!=null) {
                result.setUrl(url.asText());
            }
            if(time!=null) {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Date date = sdf.parse(time.asText());
                result.setTime(date);
            }
            return result;
        }
    }
}

启动zk

登录 hadoop1，执行下面命令：

/program/bin/zk.sh start

启动kafka

登录 hadoop1，执行下面命令：

/program/bin/kafka.sh start

启动 flink

在 idea 中运行 flink 程序

启动生产者、指定主题

登录 hadoop1，执行下面命令：

/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic kafka2flink

然后输入：

>lucy 韩梅梅
>lili lilei
>java hadoop

执行结果

在 idea 控制台中打印结果如下：

lucy ��÷÷
lili lilei
java hadoop

linux 控制台输入中文，会有问题，可忽略

参考

https://blog.csdn.net/qq_31866793/article/details/100656297

https://blog.csdn.net/qq_44962429/article/details/103731379

原文出处：/show_1IX45K305RdB.html

吉ICP备15003043号