@Override
    public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
        ElasticsearchTableInfo elasticsearchTableInfo = (ElasticsearchTableInfo) targetTableInfo;
        esTableInfo = elasticsearchTableInfo;
        clusterName = elasticsearchTableInfo.getClusterName();
        String address = elasticsearchTableInfo.getAddress();
        String[] addr = StringUtils.split(address, ",");
        esAddressList = Arrays.asList(addr);
        index = elasticsearchTableInfo.getIndex();
        type = elasticsearchTableInfo.getEsType();
        String id = elasticsearchTableInfo.getId();
        String[] idField = StringUtils.split(id, ",");
        registerTableName = elasticsearchTableInfo.getName();
        parallelism = Objects.isNull(elasticsearchTableInfo.getParallelism()) ?
                parallelism : elasticsearchTableInfo.getParallelism();

        for(int i = 0; i < idField.length; ++i) {
            if (!EsUtil.checkWhetherUsePosition(id)) {
                ids = Arrays.stream(org.apache.commons.lang.StringUtils.split(id, ",")).map(String::valueOf).collect(Collectors.toList());
            } else {
                //compatible old version of generate doc' id
                usePosition = true;
                ids = Arrays.stream(org.apache.commons.lang.StringUtils.split(id, ",")).map(Integer::valueOf).collect(Collectors.toList());
            }
        }


        columnTypes = elasticsearchTableInfo.getFieldTypes();

        return this;
    }
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