您现在的位置是：首页 >技术交流 >Seatunnel从mysql同步数据到kafka时，ISO 8601时间类型转换为Unix时间戳问题的解决网站首页技术交流

背景： 项目中使用seatunnel组件做mysql的数据同步到kafka中进行流处理，捕获其中的异常数据，但捕获到的数据发现时间类型字段做了转换由"YYYY-MM-DD HH:mm:ss.ss"格式转换为纯数字类型unix时间戳格式，到seatunnel文档中查找未发现有对应的设置项，遂对seatunnel源码在本地做了定制的开发。

项目中使用的seatunnel版本2.3.4

1. seatunnel配置文件

source {
  MySQL-CDC {
    result_table_name= "s3"
    # server-id  用于事务区
    server-id = 
    base-url = 
    username = 
    password = 
    #catalog      ={factory=MySQL}
    table-names = 
    startup.mode = 
    format = "compatible_debezium_json"
    debezium = {
        key.converter.schemas.enable = false   #include schema into kafka message
        value.converter.schemas.enable = false #
        include.schema.changes    = false         #include ddl
        database.server.name      =  "mp"           #topic prefix
        snapshot.mode             = "never"                #initial只读取最后一条记录 ,never读取断开时的最后一条记录
        decimal.handling.mode     = "string"       #decimal 类型值按double/string处理
        time.precision.mode       = "connect"        #转成毫秒
        skipped.operations        = "u"
        # 新增的配置项，用于指定序列化后的时间戳字段类型
		datetime.format.datetime  = "yyyy-MM-dd'T'HH:mm:ss:SSS'Z'"
        datetime.format.datetime.zone = "UTC"
        }
    }
}

2. 代码变动
   在org.apache.seatunnel.connectors.cdc.debezium.json路径下新增一个自定义时间格式转换类，在配置文件加载阶段，会读取datetime.format.datetime和datetime.format.datetime.zone两个配置项，如果这两个配置项有值，则会执行到convertToJsonWithoutEnvelope方法，去做时间类型转换

public class CustomDateJsonConverter extends JsonConverter {
    public CustomDateJsonConverter() {
        super();
    }
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        super.configure(configs, isKey);
    }
    private JsonNode convertToJsonWithoutEnvelope(Schema schema, Object value, String dateTimeFormat, String dateTimeZone) {
        if (value == null) {
            if (schema == null) // Any schema is valid and we don't have a default, so treat this as an optional schema
            {
                return null;
            }
            if (schema.defaultValue() != null) {
                return convertToJsonWithoutEnvelope(schema, schema.defaultValue(), dateTimeFormat, dateTimeZone);
            }
            if (schema.isOptional()) {
                return JSON_NODE_FACTORY.nullNode();
            }
            throw new DataException("Conversion error: null value for field that is required and has no default value");
        }

        if (schema != null && schema.name() != null) {
            LogicalTypeConverter logicalConverter = LOGICAL_CONVERTERS.get(schema.name());
            if (logicalConverter != null) {
                // 对日期类型做处理，转换为ISO 8601格式的字符串
                if (value instanceof Date) {
                    return dateFormatter((Date) value, dateTimeFormat, dateTimeZone);
                }
                return logicalConverter.toJson(schema, value, config);
            }
        }

        try {
            final Schema.Type schemaType;
            if (schema == null) {
                schemaType = ConnectSchema.schemaType(value.getClass());
                if (schemaType == null) {
                    throw new DataException("Java class " + value.getClass() + " does not have corresponding schema type.");
                }
            } else {
                schemaType = schema.type();
            }


            switch (schemaType) {
                case INT8:
                    return JSON_NODE_FACTORY.numberNode((Byte) value);
                case INT16:
                    return JSON_NODE_FACTORY.numberNode((Short) value);
                case INT32:
                    return JSON_NODE_FACTORY.numberNode((Integer) value);
                case INT64:
                    return JSON_NODE_FACTORY.numberNode((Long) value);
                case FLOAT32:
                    return JSON_NODE_FACTORY.numberNode((Float) value);
                case FLOAT64:
                    return JSON_NODE_FACTORY.numberNode((Double) value);
                case BOOLEAN:
                    return JSON_NODE_FACTORY.booleanNode((Boolean) value);
                case STRING:
                    CharSequence charSeq = (CharSequence) value;
                    return JSON_NODE_FACTORY.textNode(charSeq.toString());
                case BYTES:
                    if (value instanceof byte[]) {
                        return JSON_NODE_FACTORY.binaryNode((byte[]) value);
                    } else if (value instanceof ByteBuffer) {
                        return JSON_NODE_FACTORY.binaryNode(((ByteBuffer) value).array());
                    } else {
                        throw new DataException("Invalid type for bytes type: " + value.getClass());
                    }
                case ARRAY: {
                    Collection collection = (Collection) value;
                    ArrayNode list = JSON_NODE_FACTORY.arrayNode();
                    for (Object elem : collection) {
                        Schema valueSchema = schema == null ? null : schema.valueSchema();
                        JsonNode fieldValue = convertToJsonWithoutEnvelope(valueSchema, elem, dateTimeFormat, dateTimeZone);
                        list.add(fieldValue);
                    }
                    return list;
                }
                case MAP: {
                    Map<?, ?> map = (Map<?, ?>) value;
                    // If true, using string keys and JSON object; if false, using non-string keys and Array-encoding
                    boolean objectMode;
                    if (schema == null) {
                        objectMode = true;
                        for (Map.Entry<?, ?> entry : map.entrySet()) {
                            if (!(entry.getKey() instanceof String)) {
                                objectMode = false;
                                break;
                            }
                        }
                    } else {
                        objectMode = schema.keySchema().type() == Schema.Type.STRING;
                    }
                    ObjectNode obj = null;
                    ArrayNode list = null;
                    if (objectMode) {
                        obj = JSON_NODE_FACTORY.objectNode();
                    } else {
                        list = JSON_NODE_FACTORY.arrayNode();
                    }
                    for (Map.Entry<?, ?> entry : map.entrySet()) {
                        Schema keySchema = schema == null ? null : schema.keySchema();
                        Schema valueSchema = schema == null ? null : schema.valueSchema();
                        JsonNode mapKey = convertToJsonWithoutEnvelope(keySchema, entry.getKey(), dateTimeFormat, dateTimeZone);
                        JsonNode mapValue = convertToJsonWithoutEnvelope(valueSchema, entry.getValue(), dateTimeFormat, dateTimeZone);

                        if (objectMode) {
                            obj.set(mapKey.asText(), mapValue);
                        } else {
                            list.add(JSON_NODE_FACTORY.arrayNode().add(mapKey).add(mapValue));
                        }
                    }
                    return objectMode ? obj : list;
                }
                case STRUCT: {
                    Struct struct = (Struct) value;
                    if (!struct.schema().equals(schema)) {
                        throw new DataException("Mismatching schema.");
                    }
                    ObjectNode obj = JSON_NODE_FACTORY.objectNode();
                    for (Field field : schema.fields()) {
                        obj.set(field.name(), convertToJsonWithoutEnvelope(field.schema(), struct.get(field), dateTimeFormat, dateTimeZone));
                    }
                    return obj;
                }
            }

            throw new DataException("Couldn't convert " + value + " to JSON.");
        } catch (ClassCastException e) {
            String schemaTypeStr = (schema != null) ? schema.type().toString() : "unknown schema";
            throw new DataException("Invalid type for " + schemaTypeStr + ": " + value.getClass());
        }
    }

    private static TextNode dateFormatter(Date date, String dateTimeFormat, String timeZone) {
        if (StringUtils.isBlank(dateTimeFormat)) {
            // 默认的dateTime的格式化，当前支持UTC,UTC-8两种
            switch (dateTimeFormat) {
                case "UTC":
                    dateTimeFormat = DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS_SSS_UTC_ZERO.getValue();
                    break;
                case "UTC-8":
                    dateTimeFormat = DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS_SSS_UTC.getValue();
                    break;
                default:
                    dateTimeFormat = DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS_SSS_UTC_ZERO.getValue();
                    break;
            }
        }
        return JSON_NODE_FACTORY.textNode(DateTimeUtils.byInputFormatterSting(date.toInstant().atZone(ZoneId.of(timeZone)).toLocalDateTime(), dateTimeFormat));
    }
}

代码改动后打包替换掉原来的connector-cdc-mysql包，启动服务即可