private RichSinkFunction createEsSinkFunction(){

        // check whether id fields is exists in columns if not use position to generate doc's id
        if (!usePosition
                && ids != null
                && ids.size() != 0) {
            List<String> filedNamesLists = Arrays.asList(fieldNames);
            Preconditions.checkState(filedNamesLists.containsAll(ids), "elasticsearch6 type of id %s is should be exists in columns %s.", ids, filedNamesLists);
        }
        CustomerSinkFunc customerSinkFunc = new CustomerSinkFunc(index, type, Arrays.asList(fieldNames), Arrays.asList(columnTypes), ids, usePosition);

        Map<String, String> userConfig = new HashMap<>();
        userConfig.put("cluster.name", clusterName);
        // This instructs the sink to emit after every element, otherwise they would be buffered
        userConfig.put(org.apache.flink.streaming.connectors.elasticsearch5.ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "" + bulkFlushMaxActions);
        List<InetSocketAddress> transports = new ArrayList<>();

        for(String address : esAddressList){
            String[] infoArray = StringUtils.split(address, ":");
            int port = 9300;
            String host = infoArray[0];
            if(infoArray.length > 1){
                port = Integer.valueOf(infoArray[1].trim());
            }

            try {
                transports.add(new InetSocketAddress(InetAddress.getByName(host), port));
            }catch (Exception e){
                logger.error("", e);
                throw new RuntimeException(e);
            }
        }

        boolean authMesh = esTableInfo.isAuthMesh();
        if (authMesh) {
            String authPassword = esTableInfo.getUserName() + ":" + esTableInfo.getPassword();
            userConfig.put("xpack.security.user", authPassword);
        }

        return new MetricElasticsearchSink(userConfig, transports, customerSinkFunc, esTableInfo);
    }
table output elastic5plugin
Date: 2018/7/13
Company: www.dtstack.com
@author huyifan_zju@163.com
elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java