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

为自己记录一下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 操作中排除它们。

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

连接器参数

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，以上为案例所以内容。

以下是补充内容。

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

可以跳过

  <properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><scala.version>2.12</scala.version><java.version>1.8</java.version><flink.version>1.14.4</flink.version><fastjson.version>1.2.62</fastjson.version><hadoop.version>2.8.3</hadoop.version><scope.mode>compile</scope.mode><slf4j.version>1.7.30</slf4j.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-scala-bridge_${scala.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_${scala.version}</artifactId><version>${flink.version}</version><!--            本地测试时注释 scope -->
<!--          <scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_${scala.version}</artifactId><version>${flink.version}</version><!--            本地测试时注释 scope -->
<!--          <scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_${scala.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_${scala.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_${scala.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>${flink.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><!-- Add log dependencies when debugging locally --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>${slf4j.version}</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>${slf4j.version}</version></dependency><!-- flink-doris-connector --><dependency><groupId>org.apache.doris</groupId><artifactId>flink-doris-connector-1.14_2.12</artifactId><version>1.1.0</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.12</version></dependency><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>2.2.0</version><exclusions><exclusion><artifactId>flink-shaded-guava</artifactId><groupId>org.apache.flink</groupId></exclusion></exclusions></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime-web_${scala.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-statebackend-rocksdb_2.11</artifactId><version>${flink.version}</version></dependency><!--保存检查点到hdfs上--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency></dependencies>

kafka connector参数