@Override
    public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
        esTableInfo = (ElasticsearchTableInfo) targetTableInfo;
        index = esTableInfo.getIndex();
        index_definition = esTableInfo.getIndex_definition();
        columnTypes = esTableInfo.getFieldTypes();
        esAddressList = Arrays.asList(esTableInfo.getAddress().split(","));
        String id = esTableInfo.getId();
        registerTableName = esTableInfo.getName();
        parallelism = Objects.isNull(esTableInfo.getParallelism()) ? parallelism : esTableInfo.getParallelism();

        if (!StringUtils.isEmpty(id)) {
            idFiledNames = Arrays.stream(StringUtils.split(id, ",")).map(String::valueOf).collect(Collectors.toList());
        }
        return this;
    }
@description:
@program: flink.sql
@author: lany
@create: 2021/01/04 17:20
elasticsearch7/elasticsearch7-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java