flink 1.10.1 java版本elasticsearch sink写入处理结果到elasticsearch-CFANZ编程社区

本文的基础环境可以参考flink 1.10.1 java版本wordcount演示 (nc + socket)，在此基础上增加输出结果到elasticsearch。

1. 添加依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
    <version>1.10.1</version>
    <!--<scope>${scope}</scope>-->
</dependency>

2. 测试代码

package com.demo.redis;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;

/**
 * flink结果写入elasticsearch
 */
public class FlinkESSinkDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> dataStream = env.socketTextStream("192.168.0.181",9000);

        SingleOutputStreamOperator<String> flatMap = dataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] strings = value.split(" ");
                for (String s : strings) {
                    out.collect(s);
                }
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> map = flatMap.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = map.keyBy("f0").sum(1);

        DataStream<String> result = sum.map(new MapFunction<Tuple2<String, Integer>, String>() {
            @Override
            public String map(Tuple2<String, Integer> data) throws Exception {
                return data.f0 + ":" + data.f1;
            }
        });

        // 配置es参数
        ArrayList<HttpHost> httpHosts = new ArrayList<>();

        httpHosts.add(new HttpHost("localhost", 9200, "http"));

        ElasticsearchSink.Builder<String> esBuilder = new ElasticsearchSink
                .Builder<String>(httpHosts, new MyEsSinkFunction());

        esBuilder.setBulkFlushMaxActions(1);

        result.addSink(esBuilder.build());

        result.print();

        env.execute();

    }

    public static class MyEsSinkFunction implements ElasticsearchSinkFunction<String>
    {
        @Override
        public void process(String s, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {

           // 定义写入的数据
           HashMap<String, String> dataSouce = new HashMap<>();

           String timeStamp = "" + System.currentTimeMillis();
           dataSouce.put(timeStamp, s);

           // 创建es请求
            IndexRequest indexRequest = Requests.indexRequest()
                    .index("flink-es")
                    .source(dataSouce);

            try {
                requestIndexer.add(indexRequest);
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }

        }
    }


}

设置只有有记录时，立即写入ES

esBuilder.setBulkFlushMaxActions(1);