Flick batch reading es

Flink Real time consumption kafka data , The data is processed , Enrichment 、 Cleaning and other operations , write in ES. In flow calculation , This scenario is very common .
In this paper ES Batch operation of BulkProcessor The way , This method uses TransportClient, be based on Tcp agreement ; and rest The method is restClient, be based on http agreement , The accuracy of the results cannot be guaranteed .
Reference resources

One 、 Depend on the preparation ：

Mainly depends on ：

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
    <version>${elasticsearch.version}</version>
</dependency>
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>transport</artifactId>
    <version>${elasticsearch.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

The main function ：

```java
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.qianxin.ida.enrich.ElasticsearchSink;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
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 java.util.Arrays;
import java.util.Properties;
@Slf4j
public class KafkaToEs {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        try {
            env.enableCheckpointing(10000);
            //topic list 
            String[] topics = new String[]{"topic1", "topic2"};
            // many topic situation 
            Arrays.stream(topics).forEach(topic -> {
                SingleOutputStreamOperator<JSONObject> dateStream = env.addSource(new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(),
                        new Properties()).setStartFromLatest())
                        .map(new MapFunction<String, JSONObject>() {
                            // Specific data cleaning operations 
                            public JSONObject map(String value) throws Exception {
                                JSONObject jsonObject = new JSONObject();
                                jsonObject = JSON.parseObject(value);
                                return jsonObject;
                            }
                        });
                dateStream.print();
                // Customize sink
                dateStream.addSink(new ElasticsearchSink());
            });
            env.execute("kafka2es");
        } catch (Exception e) {
            log.error("kafka2es fail " + e.getMessage());
        }
    }
}

ElasticsearchSink function :

package com.qianxin.ida.enrich;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import java.net.InetAddress;
import static com.qianxin.ida.utils.Constants.INDEX_SUFFIX;
@Slf4j
public class ElasticsearchSink extends RichSinkFunction<JSONObject> implements SinkFunction<JSONObject> {
    private static BulkProcessor bulkProcessor = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        //BulkProcessor Is a thread safe batch processing class , Allows easy setup of   Refresh   A new batch request  
        Settings settings = Settings.builder()
                .put("cluster.name", "elasticsearch")
                .put("client.transport.sniff", false)
                .build();
        PreBuiltTransportClient preBuiltTransportClient = new PreBuiltTransportClient(settings);
        TransportClient client = preBuiltTransportClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("127.0.0.1"), 9300));
        BulkProcessor.Listener listener = buildListener();
        BulkProcessor.Builder bulk = BulkProcessor.builder(client, listener);
        // Set the time to refresh the new batch request according to the currently added operand ( The default value is 1000,-1 Ban )
        bulk.setBulkActions(Property.getIntValue("bulk_actions"));
        // Set the time to refresh the new batch request according to the currently added operation size ( The default is 5Mb,-1 Ban )
        bulk.setBulkSize(new ByteSizeValue(Property.getLongValue("bulk_size"), ByteSizeUnit.MB));
        // Set the number of concurrent requests allowed to execute ( The default is 1,0 Only a single request is allowed )
        bulk.setConcurrentRequests(Property.getIntValue("concurrent_request"));
        // Set a refresh interval , If the interval passes , Refresh any pending batch requests ( The default is not set )
        bulk.setFlushInterval(TimeValue.timeValueSeconds(Property.getLongValue("flush_interval")));
        // Set a constant fallback strategy , Initial wait 1 Second , Again up 3 Time 
        bulk.setBackoffPolicy(BackoffPolicy
                .constantBackoff(TimeValue.timeValueSeconds(Property.getLongValue("time_wait")),
                        Property.getIntValue("retry_times")));
        bulkProcessor = bulk.build();
        super.open(parameters);
    }
    private static BulkProcessor.Listener buildListener() throws InterruptedException {
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long l, BulkRequest bulkRequest) {
            }
            @Override
            public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
            }
            @Override
            public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
            }
        };
        return listener;
    }
    @Override
    public void invoke(JSONObject jsonObject, Context context) throws Exception {
        try {
            String topic = jsonObject.getString("topic");
            String index = "index_";
            bulkProcessor.add(new IndexRequest(index)
                    .type(topic)
                    .source(jsonObject));
        } catch (Exception e) {
            log.error("sink  error  {
   {}}, The news is {
   {}}", e.getMessage(), jsonObject);
        }
    }
}