private IndexRequest createIndexRequest(Row element) {

        Map<String, Object> dataMap = EsUtil.rowToJsonMap(element,fieldNames,fieldTypes);
        int length = Math.min(element.getArity(), fieldNames.size());
        for(int i=0; i<length; i++){
            if (element.getField(i) instanceof Date) {
                dataMap.put(fieldNames.get(i), DateUtil.transformSqlDateToUtilDate((Date) element.getField(i)));
                continue;
            }
            if (element.getField(i) instanceof Timestamp) {
                dataMap.put(fieldNames.get(i), ((Timestamp) element.getField(i)).getTime());
                continue;
            }
            dataMap.put(fieldNames.get(i), element.getField(i));
        }

        String idFieldStr = "";
        if (null != ids) {
            if (!usePosition) {
                idFieldStr = ids.stream()
                        .map(filedName -> (String) filedName)
                        .map(filedName -> dataMap.get(filedName).toString())
                        .collect(Collectors.joining(ID_VALUE_SPLIT));
            } else {
                // compatible old version of generate doc's id
                // index start at 1,
                idFieldStr = ids.stream()
                        .map(index -> (Integer) index)
                        .filter(index -> index > 0 && index <= element.getArity())
                        .map(index -> element.getField( index - 1).toString())
                        .collect(Collectors.joining(ID_VALUE_SPLIT));
            }
        }

        if (StringUtils.isEmpty(idFieldStr)) {
            return Requests.indexRequest()
                    .index(index)
                    .type(type)
                    .source(dataMap);
        }

        return Requests.indexRequest()
                .index(index)
                .type(type)
                .id(idFieldStr)
                .source(dataMap);
    }
Reason:
Date: 2017/7/19
Company: www.dtstack.com
@author xuchao
elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFunc.java