Flink从Kafka取数WordCount后TableApi写入ES

一、背景说明

需求为从Kafka消费对应主题数据，通过TableApi对数据进行WordCount后，基于DDL写法将数据写入ES。

二、代码部分

说明：代码中关于Kafka及ES的连接部分可以抽象到单独的工具类使用，这里只是一个演示的小demo，后续操作均可自行扩展，如Kakfa一般处理为json格式数据，引入fastjson等工具使用富函数进行格式处理即可。

package com.flinksql.test;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
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.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;import java.util.Properties;
import static org.apache.flink.table.api.Expressions.$;/**
* @author: Lin
* @create: 2021-06-20 10:21
* @description: 使用FlinkSQL实现从kafka读取数据计算wordcount并将数据写入ES
**/
public class FlinkTableAPI_Test {
public static void main(String[] args) throws Exception {
//1.建立环境，测试不设置CK
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); //2.获取kafka端数据
Properties prop = new Properties();
prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"BD");
DataStreamSource<String> sourceDS = env
.addSource(new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(), prop)); //3.使用flatmap转换数据到javabean，使用flatmap可以实现过滤
SingleOutputStreamOperator<Tuple2<String, Integer>> flatMapDS = sourceDS
.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
String[] split = value.split(",");
for (String s : split) {
out.collect(new Tuple2<>(s, 1));
}}
}); //4.流数据转为table
Table table = tableEnv.fromDataStream(flatMapDS);
Table table1 = table
.groupBy($("f0"))
.select($("f0").as("word"), $("f1").sum().as("num"));
tableEnv.toRetractStream(table1, Row.class).print(); //5.DDL方式建立临时表，写入datastream数据，为演示需要maxactions设置为1，默认是批量写入
tableEnv.executeSql("CREATE TABLE sensor (" +
" word STRING," +
" num BIGINT," +
" PRIMARY KEY (word) NOT ENFORCED" +
") WITH (" +
" 'connector' = 'elasticsearch-7'," +
" 'hosts' = 'http://localhost:9200'," +
" 'index' = 'test'," +
" 'sink.bulk-flush.max-actions' = '1')");
//6.数据写入
table1.executeInsert("sensor");
env.execute();
}
}


---

学习交流，有任何问题还请随时评论指出交流。