flink 写入Elasticsearch 踩坑小结

这是由于Flink的包加载机制引起的。
-- 解决方案一
原因: 类加载顺序问题，可以在flink-conf.yaml中加入
classloader.resolve-order: parent-first
Flink的默认加载是child-first。
但是用了parent-first配置有可能出现类冲突问题。解决办法只针对个别包出来，不要一律用parent-first, 配置如下：
classloader.parent-first-patterns.additional: javax.script; jdk;


-- 解决方案二
pom.xml 中将Elasticsearch相关依赖放在最开始位置。

java.;
scala.;
org.apache.flink.;
com.esotericsoftware.kryo;
org.apache.hadoop.;
javax.annotation.;
org.slf4j;
org.apache.log4j;
org.apache.logging;
org.apache.commons.logging;
ch.qos.logback;
org.xml;
javax.xml;
org.apache.xerces;
org.w3c

Suppressed: org.elasticsearch.client.ResponseException: method [POST], host [http://es1:9200], URI [/xxx/_count?ignore_unavailable=false&expand_wildcards=open&allow_no_indices=true], status line [HTTP/1.1 403 Forbidden]
{"error":{"root_cause":[{"type":"security_exception","reason":"action [indices:data/read/search] is unauthorized for user [anonymous_admin]"}],"type":"security_exception","reason":"action [indices:data/read/search] is unauthorized for user [anonymous_admin]"},"status":403}

public class ESTest2 {


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


        String user = "user";
        String pwd = "pwd";
        String auth = Base64.encodeBase64String((user+":"+pwd).getBytes());
        
        System.out.println(auth);


        RestClientBuilder clientBuilder = RestClient.builder(new HttpHost("es1", 9200));


        clientBuilder.setDefaultHeaders(new BasicHeader[]{new BasicHeader("Authorization", "Basic " + auth)});


        RestHighLevelClient client = new RestHighLevelClient(clientBuilder);


        CountRequest countRequest = new CountRequest();
        countRequest.indices(index_name);
        CountResponse countResponse = client.count(countRequest, RequestOptions.DEFAULT);
        System.out.println(countResponse.getCount());


        client.close();
    }
}

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.http.HttpHost;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;


import java.util.Map;


/**
 * @description: 自定义es sink
 * @author: mdz
 * @date: 2021/7/14
 **/
public class MyESSink extends RichSinkFunction<Map<String, String>> {
    private BulkRequest request = null;
    private RestHighLevelClient client = null;
    private CredentialsProvider credentialsProvider = null;


    private String esHost = null;
    private int esPort;
    private String esUsername = null;
    private String esPassword = null;
    private String index = null;
    private String type = null;


    public MyESSink(String esHost, int esPort, String esUsername, String esPassword, String index, String type) {
        this.esHost = esHost;
        this.esPort = esPort;
        this.esUsername = esUsername;
        this.esPassword = esPassword;
        this.index = index;
        this.type = type;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        String user = esUsername;
        String pwd = esPassword;
        String auth = Base64.encodeBase64String((user+":"+pwd).getBytes());
        RestClientBuilder clientBuilder = RestClient.builder(new HttpHost(esHost, esPort));
        clientBuilder.setDefaultHeaders(new BasicHeader[]{new BasicHeader("Authorization", "Basic " + auth)});
        client = new RestHighLevelClient(clientBuilder);
        super.open(parameters);
    }


    @Override
    public void invoke(Map<String, String> value, Context context) throws Exception {
        request = new BulkRequest();
        request.add(new IndexRequest(index,type).id(value.get("id")).source(value,XContentType.JSON));
        client.bulk(request, RequestOptions.DEFAULT);


    }




    @Override
    public void close() throws Exception {
        client.close();
        super.close();
    }
}

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
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.streaming.util.serialization.JSONKeyValueDeserializationSchema;
import java.util.*;


/**
 * @description: pulsar写入到es
 * @author: mdz
 * @date: 2021/5/19
 **/
public class Kafka2Es {


    public static void main(String[] args) throws Exception {
        //创建环境，设置参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.disableOperatorChaining();
        env.enableCheckpointing(60 * 1000 * 5, CheckpointingMode.EXACTLY_ONCE);


        //定义kafka相关参数
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "xxx:9092,xxx:9092,xxx:9092");
        properties.setProperty("group.id","gp_sdbd-etl-test07");
        properties.setProperty("request.timeout.ms", "600000");
        properties.setProperty("enable.auto.commit","false");






        //从kafka获取数据
        FlinkKafkaConsumer<Object> smartnewsSource =new FlinkKafkaConsumer("my_topic1",
                new JSONKeyValueDeserializationSchema(true),
                properties
        );
        smartnewsSource.setStartFromEarliest();   // 从最开始处消费
        DataStreamSource<Object> smartnewsDS = env.addSource(smartnewsSource);


        SingleOutputStreamOperator<Map<String, String>> mapDS = smartnewsDS.map(new MapFunction<Object, Map<String, String>>() {
            @Override
            public HashMap<String, String> map(Object value) throws Exception {
                HashMap<String, String> map1 = new HashMap<>();
                map1.put("data",value.toString());
                return map1;
            }
        });




//        //写入es
          mapDS.addSink(new MyESSink("esIp",9200,"user","pass","index","type"));
        env.execute();
    }
}