??我們實時的流處理入 ElasticSearch 中還是比較麻煩的琼懊，雖然說 flink 提供了相關(guān)的 sink 接口捂掰，但是一般來說僅僅是簡單的將數(shù)據(jù)插入而已，對于優(yōu)化以及使用用戶名和密碼登錄操作的話，不管官網(wǎng)還是網(wǎng)上票腰，寫得零零碎碎的，對于大佬來說可以拼接起來再用，但是對于像我這種菜鳥來說，那簡直是看天書一樣射富，一愣一愣的。今天寫這個案例主要是項目中涉及了這個需求粥帚，廢了半條命終于整理出來了胰耗，現(xiàn)在做個總結(jié)，以便避免初學(xué)者再掉坑芒涡。
??廢話不多說柴灯，接下來我們開搞。费尽。赠群。

一、啟動服務(wù)器

[syy@nfdw elasticsearch-7.6.1]$ pwd
/opt/modules/elasticsearch-7.6.1
[syy@nfdw elasticsearch-7.6.1]$ bin/elasticsearch

[syy@nfdw kibana-7.6.1-linux-x86_64]$ pwd
/opt/modules/kibana-7.6.1-linux-x86_64
[syy@nfdw kibana-7.6.1-linux-x86_64]$ bin/kibana

登錄 kibana 控制臺：http://IP:5601/app/kibana#/dev_tools/console ,登錄成功如下：

image.png

二旱幼、代碼實現(xiàn)

2.1 添加依賴

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch7 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/com.google.code.gson/gson -->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.6</version>
        </dependency>

2.2 主體代碼

public class App {

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

        //  獲取環(huán)境對象
        StreamExecutionEnvironment env = GetStreamExecutionEnvironment.getEnv();
        //請求kafka數(shù)據(jù)
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers","cdh101:9092");
        prop.setProperty("group.id","cloudera_mirrormaker");
        prop.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        FlinkKafkaConsumer011<String> myConsumer = new FlinkKafkaConsumer011("luchangyin", new SimpleStringSchema() ,prop);
        myConsumer.setStartFromLatest();  //最近的

        //請求kafka數(shù)據(jù)
        DataStreamSource<String> dataStream = env.addSource(myConsumer);
        //dataStream.print();   // {"id":"226","name":"tang tang - 226","sal":280751,"dept":"美女部","ts":1615191802523}

        SingleOutputStreamOperator<Employees> result = dataStream.map(new MapFunction<String, Employees>() {

            @Override
            public Employees map(String s) throws Exception {
                Employees emp = MyJsonUtils.str2JsonObj(s);
                emp.setEmpStartTime(new Date(emp.getTs()));
                emp.setDt(MyDateUtils.getDate2Second(emp.getEmpStartTime()));
                return emp;
            }
        });

        //result.print();
        // Employees(eId=257, eName=fei fei - 257, eSal=97674.0, eDept=美女部, ts=1615251002894, empStartTime=Tue Mar 09 08:50:02 GMT+08:00 2021, dt=2021-03-09)

        // 設(shè)置ES的服務(wù)器地址
        List<HttpHost> esAddresses = null;
        try {
            esAddresses = ESSinkUtil.getEsAddresses("10.122.1.115:9200");
        } catch (MalformedURLException e) {
            e.printStackTrace();
        }

        // 我們可以通過調(diào)試此方法的三個數(shù)值參數(shù)進(jìn)行優(yōu)化
        ESSinkUtil.addSink(esAddresses, "elastic", "123456", 100,100, 1,
                5, result, new ElasticsearchSinkFunction<Employees>() {
                    @Override
                    public void process(Employees emp, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                        String indexStr = "employee_"+ MyDateUtils.getTime2Day(emp.getEmpStartTime()).replaceAll("-","");
                        System.out.println("索引為-> "+ indexStr);
                        requestIndexer.add(Requests.indexRequest()
                                .index(indexStr)
                                .source(GsonUtil.toJSONBytes(emp), XContentType.JSON));
                    }
                });


        env.execute("wo xi huan ni");

    }

}

2.3 實現(xiàn)SinkEs的工具類：

package com.nfdw.utils;

import org.apache.commons.lang.StringUtils;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;

import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;

public class ESSinkUtil {

    /**
     * es sink
     *
     * @param hosts               es hosts
     * @param bulkFlushMaxActions bulk flush size
     * @param parallelism         并行數(shù)
     * @param data                數(shù)據(jù)
     * @param func
     * @param <T>
     */
    public static <T> void addSink(List<HttpHost> esAddresses, String userName, String passwd, int bulkFlushMaxActions,
                                   int bulkFlushMaxSizeMb, long bulkFlushInterval, int parallelism,
                                   SingleOutputStreamOperator<T> data, ElasticsearchSinkFunction<T> func) {
        //todo:xpack security
        ElasticsearchSink.Builder<T> esSinkBuilder = new ElasticsearchSink.Builder<>(esAddresses, func);

        // 鑒權(quán)查描，正對寫 es 需要密碼的場景
        if(StringUtils.isNotEmpty(userName) && StringUtils.isNotEmpty(passwd)){
            esSinkBuilder.setRestClientFactory(new HDRestClientFactory(userName,passwd));
        }

        //失敗處理策略
        esSinkBuilder.setFailureHandler(new RetryRequestFailureHandler());

        //bulk
        esSinkBuilder.setBulkFlushMaxActions(bulkFlushMaxActions);
        esSinkBuilder.setBulkFlushMaxSizeMb(bulkFlushMaxSizeMb);
        esSinkBuilder.setBulkFlushInterval(bulkFlushInterval);

        //-----------------------------------
        data.addSink(esSinkBuilder.build()).setParallelism(parallelism);
    }

    /**
     * 解析配置文件的 es hosts
     *
     * @param hosts
     * @return
     * @throws MalformedURLException
     */
    public static List<HttpHost> getEsAddresses(String hosts) throws MalformedURLException {
        String[] hostList = hosts.split(",");
        List<HttpHost> addresses = new ArrayList<>();
        for (String host : hostList) {
            if (host.startsWith("http")) {
                URL url = new URL(host);
                addresses.add(new HttpHost(url.getHost(), url.getPort()));
            } else {
                String[] parts = host.split(":", 2);
                if (parts.length > 1) {
                    addresses.add(new HttpHost(parts[0], Integer.parseInt(parts[1])));
                } else {
                    throw new MalformedURLException("invalid elasticsearch hosts format");
                }
            }
        }
        return addresses;
    }

}

2.4 設(shè)置密碼操作類 HDRestClientFactory:

package com.nfdw.utils;

import org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClientBuilder;

public class HDRestClientFactory implements RestClientFactory {

    private String userName;
    private String password;
    transient CredentialsProvider credentialsProvider;

    public HDRestClientFactory(String userName, String password) {
        this.userName = userName;
        this.password = password;
    }

    @Override
    public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
        if (credentialsProvider == null) {
            credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(userName, password));
        }
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        }).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder builder) {
                builder.setConnectTimeout(5000);
                builder.setSocketTimeout(60000);
                builder.setConnectionRequestTimeout(2000);
                return builder;
            }
        });
    }
}

2.5 創(chuàng)建失敗策列處理類 RetryRequestFailureHandler :

package com.nfdw.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.util.ExceptionUtils;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
import java.io.IOException;
import java.net.SocketTimeoutException;
import java.util.Optional;

@Slf4j
public class RetryRequestFailureHandler implements ActionRequestFailureHandler {

    public RetryRequestFailureHandler() {
    }

    @Override
    public void onFailure(ActionRequest actionRequest, Throwable throwable, int i, RequestIndexer requestIndexer) throws Throwable {
        if (ExceptionUtils.findThrowable(throwable, EsRejectedExecutionException.class).isPresent()) {
            requestIndexer.add(new ActionRequest[]{actionRequest});
        } else {
            if (ExceptionUtils.findThrowable(throwable, SocketTimeoutException.class).isPresent()) {
                return;
            } else {
                Optional<IOException> exp = ExceptionUtils.findThrowable(throwable, IOException.class);
                if (exp.isPresent()) {
                    IOException ioExp = exp.get();
                    if (ioExp != null && ioExp.getMessage() != null && ioExp.getMessage().contains("max retry timeout")) {
                        log.error(ioExp.getMessage());
                        return;
                    }
                }
            }
            throw throwable;
        }
    }
}

2.6 創(chuàng)建一個 gson 解析類：

package com.nfdw.utils;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import java.lang.reflect.Type;
import java.nio.charset.Charset;

public class GsonUtil {

    private final static Gson gson = new Gson();

    private final static Gson disableHtmlEscapingGson = new GsonBuilder().disableHtmlEscaping().create();

    public static <T> T fromJson(String value, Class<T> type) {
        return gson.fromJson(value, type);
    }

    public static <T> T fromJson(String value, Type type) {
        return gson.fromJson(value, type);
    }

    public static String toJson(Object value) {
        return gson.toJson(value);
    }

    public static String toJsonDisableHtmlEscaping(Object value) {
        return disableHtmlEscapingGson.toJson(value);
    }

    public static byte[] toJSONBytes(Object value) {
        return gson.toJson(value).getBytes(Charset.forName("UTF-8"));
    }

}

三、運行程序結(jié)果查詢?nèi)缦?/h6>

image.png

??這里需要注意的一點是：sinkEs 的流必須是 SingleOutputStreamOperator 的對象速警，至于優(yōu)化就是調(diào)節(jié)工具類中的那幾個數(shù)值參數(shù)即可叹誉，好了，F(xiàn)link 對 ES 的操作到此為止闷旧，希望能夠幫助到你哦长豁。。忙灼。