@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, ",");
        idIndexList = new ArrayList<>();
        registerTableName = elasticsearchTableInfo.getName();
        parallelism = Objects.isNull(elasticsearchTableInfo.getParallelism()) ?
                parallelism : elasticsearchTableInfo.getParallelism();

        //kerberos
        principal = elasticsearchTableInfo.getPrincipal();
        keytab = elasticsearchTableInfo.getKeytab();
        krb5conf = elasticsearchTableInfo.getKrb5conf();
        enableKrb = elasticsearchTableInfo.isEnableKrb();

        for(int i = 0; i < idField.length; ++i) {
            idIndexList.add(Integer.valueOf(idField[i]));
        }

        columnTypes = elasticsearchTableInfo.getFieldTypes();

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