@Override
    public EsIndexRequest parse(EsSinkContext context, ProfileEvent event) {
        String uid = event.getUid();
        EsIdConfig idConfig = context.getIdConfig(uid);
        if (idConfig == null) {
            context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
            return null;
        }
        // parse fields
        String delimeter = idConfig.getSeparator();
        char cDelimeter = delimeter.charAt(0);
        String strContext = null;
        // for tab separator
        byte[] bodyBytes = event.getBody();
        int msgLength = event.getBody().length;
        int contentOffset = idConfig.getContentOffset();
        if (contentOffset > 0 && msgLength >= 1) {
            strContext = new String(bodyBytes, contentOffset, msgLength - contentOffset, Charset.defaultCharset());
        } else {
            strContext = new String(bodyBytes, Charset.defaultCharset());
        }
        // unescape
        List<String> columnVlues = UnescapeHelper.toFiledList(strContext, cDelimeter);
        int valueLength = columnVlues.size();
        List<String> fieldList = idConfig.getFieldList();
        int columnLength = fieldList.size();
        // field offset
        int fieldOffset = idConfig.getFieldOffset();
        // get field value
        Map<String, String> fieldMap = new HashMap<>();
        for (int i = fieldOffset; i < columnLength; ++i) {
            String fieldName = fieldList.get(i);
            int columnIndex = i - fieldOffset;
            String fieldValue = columnIndex < valueLength ? columnVlues.get(columnIndex) : "";
            byte[] fieldBytes = fieldValue.getBytes(Charset.defaultCharset());
            if (fieldBytes.length > context.getKeywordMaxLength()) {
                fieldValue = new String(fieldBytes, 0, context.getKeywordMaxLength());
            }
            fieldMap.put(fieldName, fieldValue);
        }

        // ftime
        String ftime = dateFormat.format(new Date(event.getRawLogTime()));
        fieldMap.put("ftime", ftime);
        // extinfo
        String extinfo = getExtInfo(event);
        fieldMap.put("extinfo", extinfo);
        String indexName = idConfig.parseIndexName(event.getRawLogTime());
        // build
        EsIndexRequest indexRequest = new EsIndexRequest(indexName, event);
        if (context.isUseIndexId()) {
            String esIndexId = uid + delimeter + event.getRawLogTime() + delimeter + esIndexIndex.incrementAndGet();
            indexRequest.id(esIndexId);
        }
        indexRequest.source(fieldMap);
        return indexRequest;
    }
parse

@param  context
@param  event
@return
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/DefaultEvent2IndexRequestHandler.java