private RichSinkFunction createEsSinkFunction(){

        // elasticsearch client settings
        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(ElasticsearchSinkBase.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "" + bulkFlushMaxActions);

        // elasticsearch hosts
        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);
            }
        }

        CustomerSinkFunc customerSinkFunc = new CustomerSinkFunc(index, type, Arrays.asList(fieldNames), Arrays.asList(columnTypes), idIndexList);

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