FlinkSQL_1.12_用连接器实现TableAPI(Sink ES)把端口数据写入ES并实现去重写入_flink sql 写入es

Flink用连接器实现TableAPI(Sink ES)把端口数据写入ES并实现去重写入

package com.atguigu.day10;
 
import com.atguigu.bean.WaterSensor;
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.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Elasticsearch;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.kafka.clients.producer.ProducerConfig;
 
import static org.apache.flink.table.api.Expressions.$;
 
/**
 * @author dxy
 * @date 2021/4/21 13:46
 */
public class FlinkSQL16_Sink_ES {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 
        //2.读取端口数据并封装成JavaBean
        DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop102", 9998);
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = socketTextStream.map(line -> {
            String[] fields = line.split(",");
            return new WaterSensor(fields[0],
                    Long.parseLong(fields[1]),
                    Double.parseDouble(fields[2]));
        });
 
        //3.使用TabelAPI
        Table sensorTable = tableEnv.fromDataStream(waterSensorDS);
 
        Table resultTable = sensorTable
                .where($("id").isEqual("ws_001"))
                .select($("id"), $("ts"), $("vc"));
 
        //4.将selectTable写入ES
        tableEnv.connect(new Elasticsearch()
                .index("sensor_sql")
                .documentType("_doc")
                .version("6")
                .host("hadoop102", 9200, "http")
                .bulkFlushMaxActions(1))
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("ts", DataTypes.BIGINT())
                        .field("vc", DataTypes.DOUBLE()))
                .withFormat(new Json())
                .inAppendMode()
                .createTemporaryTable("sensor");
 
        //6.将数据通过连接器写入文件
        resultTable.executeInsert("sensor");//Sink
 
        //7.启动
        env.execute();
    }
}

参数说明：bulkFlushMaxActions(1)每一条数据就执行写入，当然我们也可以以设置间隔多长时间进行写入，可以设置不同的参数方式实现写入。

在企业中肯定是批量写入，不是每一条就写入，ES是json格式因此我们指定的格式必须是JSON格式，其实在ES中它支持Append模式和upsert模式

在ES中append模式DOC的ID是随机给的，因为它没办法判断前面ID是追加模式。

导入ES依赖

  <dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
  <version>1.12.0</version>
</dependency>

启动es.sh

#查看节点信息
GET _cat/nodes?v

#查看索引
GET _cat/indices

#查询别名列表
GET  _cat/aliases?v

#查看系统中已有的模板清单
GET  _cat/templates

#查看系统健康状态
GET _cat/health?v

开启9998端口然后启动程序

输入数据进行测试

在ES中查索引

然后查看数据

测试成功，成功将端口数据写入ES，并且可以去重，去重自己测试！