FlinkSQL kafka完整案例可直接复制使用

guduadmin13小时前

为自己记录一下flinksql 消费kafka json数据并写入doris的完整案例

用完发现，flinksql 是真的香。

虽然尽量追求完整，但是从kafka造数据开始写，过于累赘因此省略。正文开始。

单表

kafka原始数据

{"id":1,"name":"nick","age":7,"address":"shanghai"}

原始数据形式

flinksql 连接

准备连接sql

    public static String kafkaTablePerson = "CREATE TABLE person (\n" +
            " id INT,\n" +
            " name STRING,\n" +
            " age INT,\n" +
            " address STRING\n" +
            ") WITH (\n" +
            " 'connector' = 'kafka',\n" +
            " 'topic' = 'flink',\n" +
            " 'properties.bootstrap.servers' = '172.8.10.xxx:6667',\n" +
            " 'format' = 'json',\n" +
            " 'scan.startup.mode' = 'earliest-offset'\n" +
            ")\n";

以下的连接器元数据可以在表定义中通过元数据列的形式获取。

R/W 列定义了一个元数据是可读的（R）还是可写的（W）。只读列必须声明为 VIRTUAL 以在 INSERT INTO 操作中排除它们。

键	数据类型	描述	R/W
topic	STRING NOT NULL	Kafka 记录的 Topic 名。	R
partition	INT NOT NULL	Kafka 记录的 partition ID。	R
headers	MAP NOT NULL	二进制 Map 类型的 Kafka 记录头（Header）。	R/W
leader-epoch	INT NULL	Kafka 记录的 Leader epoch（如果可用）。	R
offset	BIGINT NOT NULL	Kafka 记录在 partition 中的 offset。	R
timestamp	TIMESTAMP_LTZ(3) NOT NULL	Kafka 记录的时间戳。	R/W
timestamp-type	STRING NOT NULL	Kafka 记录的时间戳类型。可能的类型有 “NoTimestampType”， “CreateTime”（会在写入元数据时设置），或 “LogAppendTime”。	R

以下是kafka connector的参数，只写了常用的，文末有所有参数

连接器参数

参数	是否必选	默认值	数据类型	描述
connector	必选	（无）	String	指定使用的连接器，Kafka 连接器使用 ‘kafka’。
topic	required for sink	（无）	String	当表用作 source 时读取数据的 topic 名。亦支持用分号间隔的 topic 列表，如 ‘topic-1;topic-2’。注意，对 source 表而言，’topic’ 和 ‘topic-pattern’ 两个选项只能使用其中一个。当表被用作 sink 时，该配置表示写入的 topic 名。注意 sink 表不支持 topic 列表。
properties.bootstrap.servers	必选	（无）	String	逗号分隔的 Kafka broker 列表。
properties.group.id	对 source 可选，不适用于 sink	（无）	String	Kafka source 的消费组 id。如果未指定消费组 ID，则会使用自动生成的 “KafkaSource-{tableIdentifier}” 作为消费组 ID。
format	必选	（无）	String	用来序列化或反序列化 Kafka 消息的格式。请参阅格式页面以获取更多关于格式的细节和相关配置项。注意：该配置项和 ‘value.format’ 二者必需其一。
scan.startup.mode	可选	group-offsets	String	Kafka consumer 的启动模式。有效值为：‘earliest-offset’，‘latest-offset’，‘group-offsets’，‘timestamp’ 和 ‘specific-offsets’。

flinksql查询

    public static String selectALLPerson = "select id,name,age,address from person ";

主体方法

  		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        tEnv.executeSql(CreateTableSQL.kafkaTableInfo);
    	tEnv.executeSql(SelectSQL.selectALLPerson).print();

简单的几行代码就可以搞定了，和写sql 没什么区别，十分便捷。

sql筛选语句

把sql稍微复杂一点点

public static String selectPerCountNamePrint = " select name, COUNT(*) as num FROM person GROUP BY name ";

依旧可以，不过有个坑，别名不能和方法名相同，这个在其他地方似乎没有这个限制的，我试了count(*) as count 就会出现错误。

数据落盘

仅仅只是select 是不会满足使用的，需要落盘。

我这里使用的是sink doris中

    public static final String SinkDorisPerCountName = "CREATE TABLE  percountname_sink (" +
            "name STRING, \n" +
            "num BIGINT  \n" +
            ")"+
            "WITH ( \n" +
            " 'connector' = 'doris' , \n" +
            " 'fenodes' = '172.8.10.xxx:8030' , \n" +
            " 'table.identifier' = 'test_db.PerCountName' ,\n " +
            " 'username' = 'username', \n " +
            " 'password' = 'password' ,\n " +
            "  'sink.label-prefix' ='" + label + "',\n" +
            "  'sink.properties.format' = 'json',\n" +
            "  'sink.properties.read_json_by_line' = 'true'\n" +
            ")";

需要注意的是label 必须是唯一的，为了避免多次测试时遇到问题，这里写了个随机的label，确保每次都不一样。

案例中的WITH 参数基本满足使用，更多细节推荐去官网看看。

为了方便我把doris建表贴出来

CREATE TABLE IF NOT EXISTS PerCountName
(
    `name` VARCHAR(50) NOT NULL COMMENT "",
    `num` INT NOT NULL COMMENT "出现的次数"
)
UNIQUE KEY(`name`)
DISTRIBUTED BY HASH(`name`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);

insertSQL语句

public static String selectPerCountNamePrint = " select name, COUNT(*) as num FROM person GROUP BY name ";

执行

现在只需要把上文的主体方法稍微改一下即可

 		tEnv.executeSql(CreateTableSQL.kafkaTablePerson);
        tEnv.executeSql(SinkDorisTableSQL.SinkDorisPerCountName);
        tEnv.executeSql(SelectSQL.selectPerCountNameInsert);

执行的时候会发现doris里并没有数据，这是因为

为了保证Flink的Exactly Once语义，Flink Doris Connector 默认开启两阶段提交，Doris在1.1版本后默认开启两阶段提交。1.0可通过修改BE参数开启。

因此需要开启checkpoint 或者修改doris BE参数

 env.enableCheckpointing(10000);
 env.setParallelism(1);

加上之后即可

还去试了一下on yarn 是没有问题的，不过会遇到报错

可以看这篇文章

https://blog.csdn.net/weixin_45399602/article/details/127526911

好了，现在单表查询和落盘已经会了，想试一下多表查询有没有问题呢

多表join

新表结构这样

{"id":0,"name":"jack","score":141,"date":"2023-01-02"}

目标为：每个name的最大score。

简单一点直接贴了

    public static String kafkaTableInfo = "CREATE TABLE info (\n" +
            " id INT,\n" +
            " name STRING,\n" +
            " score INT,\n" +
            " `date` DATE\n" +
            ") WITH (\n" +
            " 'connector' = 'kafka',\n" +
            " 'topic' = 'flinkinfo',\n" +
            " 'properties.bootstrap.servers' = '172.8.10.xxx:6667',\n" +
            " 'format' = 'json',\n" +
            " 'scan.startup.mode' = 'earliest-offset'\n" +
            ")\n";
 public static String selectPerMaxScore = " SELECT info.name, MAX(info.score) AS max_score FROM info JOIN person ON person.name = info.name GROUP BY info.name";

join查询

tEnv.executeSql(CreateTableSQL.kafkaTableInfo);
tEnv.executeSql(CreateTableSQL.kafkaTablePerson);
tEnv.executeSql(SelectSQL.selectPerMaxScore).print();

或者写到doris里也是可以的，效果会比较明显。

总结

flinksql 消费kafka整体来说是十分简单好用的，使用时门口较低，实时性又高，可以经常使用。

OK，以上为案例所以内容。

以下是补充内容。

整个项目所需要的所有依赖

可以跳过

  
    UTF-8
    8
    8
    2.12
    1.8
    1.14.4
    1.2.62
    2.8.3
    compile
    1.7.30
  
  
      
          org.apache.flink
          flink-table-api-scala-bridge_${scala.version}
          ${flink.version}
          provided
      
      
          org.apache.flink
          flink-table-planner_${scala.version}
          ${flink.version}
          

      
      
          org.apache.flink
          flink-streaming-scala_${scala.version}
          ${flink.version}
          

      
      
          org.apache.flink
          flink-clients_${scala.version}
          ${flink.version}
      
      
          org.apache.flink
          flink-connector-jdbc_${scala.version}
          ${flink.version}
      
      
          org.apache.flink
          flink-connector-kafka_${scala.version}
          ${flink.version}
      
      
          org.apache.flink
          flink-json
          ${flink.version}
      
      
          com.alibaba
          fastjson
          ${fastjson.version}
      
      
      
          org.slf4j
          slf4j-api
          ${slf4j.version}
      
      
          org.slf4j
          slf4j-log4j12
          ${slf4j.version}
      
      
      
          org.apache.doris
          flink-doris-connector-1.14_2.12
          1.1.0
      
      
          mysql
          mysql-connector-java
          8.0.12
      
      
          com.ververica
          flink-connector-mysql-cdc
          2.2.0
          
              
                  flink-shaded-guava
                  org.apache.flink
              
          
      
      
          org.apache.flink
          flink-runtime-web_${scala.version}
          ${flink.version}
      
      
          org.apache.flink
          flink-statebackend-rocksdb_2.11
          ${flink.version}
      
      
      
          org.apache.hadoop
          hadoop-client
          ${hadoop.version}

kafka connector参数

参数	是否必选	默认值	数据类型	描述
connector	必选	（无）	String	指定使用的连接器，Kafka 连接器使用 ‘kafka’。
topic	required for sink	（无）	String	当表用作 source 时读取数据的 topic 名。亦支持用分号间隔的 topic 列表，如 ‘topic-1;topic-2’。注意，对 source 表而言，’topic’ 和 ‘topic-pattern’ 两个选项只能使用其中一个。当表被用作 sink 时，该配置表示写入的 topic 名。注意 sink 表不支持 topic 列表。
topic-pattern	可选	（无）	String	匹配读取 topic 名称的正则表达式。在作业开始运行时，所有匹配该正则表达式的 topic 都将被 Kafka consumer 订阅。注意，对 source 表而言，’topic’ 和 ‘topic-pattern’ 两个选项只能使用其中一个。
properties.bootstrap.servers	必选	（无）	String	逗号分隔的 Kafka broker 列表。
properties.group.id	对 source 可选，不适用于 sink	（无）	String	Kafka source 的消费组 id。如果未指定消费组 ID，则会使用自动生成的 “KafkaSource-{tableIdentifier}” 作为消费组 ID。
properties.*	可选	（无）	String	可以设置和传递任意 Kafka 的配置项。后缀名必须匹配在 Kafka 配置文档中定义的配置键。Flink 将移除 “properties.” 配置键前缀并将变换后的配置键和值传入底层的 Kafka 客户端。例如，你可以通过 ‘properties.allow.auto.create.topics’ = ‘false’ 来禁用 topic 的自动创建。但是某些配置项不支持进行配置，因为 Flink 会覆盖这些配置，例如 ‘key.deserializer’ 和 ‘value.deserializer’。
format	必选	（无）	String	用来序列化或反序列化 Kafka 消息的格式。请参阅格式页面以获取更多关于格式的细节和相关配置项。注意：该配置项和 ‘value.format’ 二者必需其一。
key.format	可选	（无）	String	用来序列化和反序列化 Kafka 消息键（Key）的格式。请参阅格式页面以获取更多关于格式的细节和相关配置项。注意：如果定义了键格式，则配置项 ‘key.fields’ 也是必需的。否则 Kafka 记录将使用空值作为键。
key.fields	可选	[]	List	表结构中用来配置消息键（Key）格式数据类型的字段列表。默认情况下该列表为空，因此消息键没有定义。列表格式为 ‘field1;field2’。
key.fields-prefix	可选	（无）	String	为所有消息键（Key）格式字段指定自定义前缀，以避免与消息体（Value）格式字段重名。默认情况下前缀为空。如果定义了前缀，表结构和配置项 ‘key.fields’ 都需要使用带前缀的名称。当构建消息键格式字段时，前缀会被移除，消息键格式将会使用无前缀的名称。请注意该配置项要求必须将 ‘value.fields-include’ 配置为 ‘EXCEPT_KEY’。
value.format	必选	（无）	String	序列化和反序列化 Kafka 消息体时使用的格式。请参阅格式页面以获取更多关于格式的细节和相关配置项。注意：该配置项和 ‘format’ 二者必需其一。
value.fields-include	可选	ALL	枚举类型可选值：[ALL, EXCEPT_KEY]	定义消息体（Value）格式如何处理消息键（Key）字段的策略。默认情况下，表结构中 ‘ALL’ 即所有的字段都会包含在消息体格式中，即消息键字段在消息键和消息体格式中都会出现。
scan.startup.mode	可选	group-offsets	String	Kafka consumer 的启动模式。有效值为：‘earliest-offset’，‘latest-offset’，‘group-offsets’，‘timestamp’ 和 ‘specific-offsets’。请参阅下方起始消费位点以获取更多细节。
scan.startup.specific-offsets	可选	（无）	String	在使用 ‘specific-offsets’ 启动模式时为每个 partition 指定 offset，例如 ‘partition:0,offset:42;partition:1,offset:300’。
scan.startup.timestamp-millis	可选	（无）	Long	在使用 ‘timestamp’ 启动模式时指定启动的时间戳（单位毫秒）。
scan.topic-partition-discovery.interval	可选	（无）	Duration	Consumer 定期探测动态创建的 Kafka topic 和 partition 的时间间隔。
sink.partitioner	可选	‘default’	String	Flink partition 到 Kafka partition 的分区映射关系，可选值有：default：使用 Kafka 默认的分区器对消息进行分区。fixed：每个 Flink partition 最终对应最多一个 Kafka partition。round-robin：Flink partition 按轮循（round-robin）的模式对应到 Kafka partition。只有当未指定消息的消息键时生效。自定义 FlinkKafkaPartitioner 的子类：例如 ‘org.mycompany.MyPartitioner’。请参阅下方 Sink 分区以获取更多细节。
sink.semantic	可选	at-least-once	String	定义 Kafka sink 的语义。有效值为 ‘at-least-once’，‘exactly-once’ 和 ‘none’。请参阅一致性保证以获取更多细节。
sink.parallelism	可选	（无）	Integer	定义 Kafka sink 算子的并行度。默认情况下，并行度由框架定义为与上游串联的算子相同。

db标签

网友评论

搜索: Search

最新文章

热门文章

FlinkSQL kafka完整案例 可直接复制使用

单表