位置：首页-资讯-数据库

FlinkCDC时间问题timestamp等

2023-09-09 19:52

短信预约 -IT技能 免费直播动态提醒

1.采集时间变成时间戳问题

1.1针对时间类型转换

使用flinkcdc V2.3.0（及其以下版本）采集时间，源库中为yyyy-MM-dd 类型采集后变为时间戳

源库数据样式：

采集后的json数据

{"before":{"id":"4","claim_amount":0.0,"follow_result_type":"2","logistics_settle_flag":null,"submit_operater":"s","apply_operater":"s","submit_time":1665741204000,"apply_time":1665741210000,"claim_remark":"自动提报调度：异常售后自动提报为(实物追回)。","exception_type":"AG_RETURN_LOSS","apply_id":"df1bbf781000472b8988240a55f9fb73","apply_no":"1","order_sub_no":"1","out_order_id":"1","follow_status":"PROCESSED","follow_source":"oms","loss_reason":null,"responsibility_ascription":"None","trading_no":null,"dept_code":null,"update_time":1665742789000,"create_time":1665741337000,"total_ex_goods_num":0},"after":{"id":"4","claim_amount":0.0,"follow_result_type":"2","logistics_settle_flag":null,"submit_operater":"s","apply_operater":"s","submit_time":null,"apply_time":1665741210000,"claim_remark":"自动提报调度：异常售后自动提报为(实物追回)。","exception_type":"AG_RETURN_LOSS","apply_id":"df1bbf781000472b8988240a55f9fb73","apply_no":"1","order_sub_no":"1","out_order_id":"1","follow_status":"PROCESSED","follow_source":"oms","loss_reason":null,"responsibility_ascription":"None","trading_no":null,"dept_code":null,"update_time":1668183700000,"create_time":1665741337000,"total_ex_goods_num":0},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1668154900000,"snapshot":"false","db":"flink_cdc_timestamp","sequence":null,"table":"table_flink_cdc_test1_default_problem","server_id":1,"gtid":null,"file":"","pos":509,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1668154900186,"transaction":null}

1.2解决办法：

1.2.1编写处理时间类：

import io.debezium.spi.converter.CustomConverter;import io.debezium.spi.converter.RelationalColumn;import org.apache.kafka.connect.data.SchemaBuilder;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.time.*;import java.time.format.DateTimeFormatter;import java.util.Properties;import java.util.function.Consumer;public class MySqlDateTimeConverter implements CustomConverter {    private final static Logger logger = LoggerFactory.getLogger(MySqlDateTimeConverter.class);    private DateTimeFormatter dateFormatter = DateTimeFormatter.ISO_DATE;    private DateTimeFormatter timeFormatter = DateTimeFormatter.ISO_TIME;    private DateTimeFormatter datetimeFormatter = DateTimeFormatter.ISO_DATE_TIME;    private DateTimeFormatter timestampFormatter = DateTimeFormatter.ISO_DATE_TIME;    private ZoneId timestampZoneId = ZoneId.systemDefault();    @Override    public void configure(Properties props) {        readProps(props, "format.date", p -> dateFormatter = DateTimeFormatter.ofPattern(p));        readProps(props, "format.time", p -> timeFormatter = DateTimeFormatter.ofPattern(p));        readProps(props, "format.datetime", p -> datetimeFormatter = DateTimeFormatter.ofPattern(p));        readProps(props, "format.timestamp", p -> timestampFormatter = DateTimeFormatter.ofPattern(p));        readProps(props, "format.timestamp.zone", z -> timestampZoneId = ZoneId.of(z));    }    private void readProps(Properties properties, String settingKey, Consumer callback) {        String settingValue = (String) properties.get(settingKey);        if (settingValue == null || settingValue.length() == 0) {            return;        }        try {            callback.accept(settingValue.trim());        } catch (IllegalArgumentException | DateTimeException e) {            logger.error("The {} setting is illegal: {}",settingKey,settingValue);            throw e;        }    }    @Override    public void converterFor(RelationalColumn column, ConverterRegistration registration) {        String sqlType = column.typeName().toUpperCase();        SchemaBuilder schemaBuilder = null;        Converter converter = null;        if ("DATE".equals(sqlType)) {            schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.date.string");            converter = this::convertDate;        }        if ("TIME".equals(sqlType)) {            schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.time.string");            converter = this::convertTime;        }        if ("DATETIME".equals(sqlType)) {            schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.datetime.string");            converter = this::convertDateTime;        }        if ("TIMESTAMP".equals(sqlType)) {            schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.timestamp.string");            converter = this::convertTimestamp;        }        if (schemaBuilder != null) {            registration.register(schemaBuilder, converter);        }    }    private String convertDate(Object input) {        if (input instanceof LocalDate) {            return dateFormatter.format((LocalDate) input);        }        if (input instanceof Integer) {            LocalDate date = LocalDate.ofEpochDay((Integer) input);            return dateFormatter.format(date);        }        return null;    }    private String convertTime(Object input) {        if (input instanceof Duration) {            Duration duration = (Duration) input;            long seconds = duration.getSeconds();            int nano = duration.getNano();            LocalTime time = LocalTime.ofSecondOfDay(seconds).withNano(nano);            return timeFormatter.format(time);        }        return null;    }    private String convertDateTime(Object input) {        if (input instanceof LocalDateTime) {            return datetimeFormatter.format((LocalDateTime) input);        }        return null;    }    private String convertTimestamp(Object input) {        if (input instanceof ZonedDateTime) {            // mysql的timestamp会转成UTC存储，这里的zonedDatetime都是UTC时间            ZonedDateTime zonedDateTime = (ZonedDateTime) input;            LocalDateTime localDateTime = zonedDateTime.withZoneSameInstant(timestampZoneId).toLocalDateTime();            return timestampFormatter.format(localDateTime);        }        return null;    }}

2在主方法中引入：

package com.belle.dc.stram;import com.ververica.cdc.connectors.mysql.table.StartupOptions;import org.apache.flink.api.common.eventtime.WatermarkStrategy;import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;import com.ververica.cdc.connectors.mysql.source.MySqlSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Properties;// 测试flink cdc2.3.0 对多库多表和指定时间戳的支持问题public class FlinkCdcApp {    public static String HOST = "localhost";    public static int PORT = 3306 ;    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        MySqlSource mySqlSource = MySqlSource.builder()                .hostname(HOST)                .port(PORT)                .databaseList("flink_cdc_timestamp","magic_api") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".                .tableList("flink_cdc_timestamp.table_flink_cdc_test1_default_problem","magic_api.magic_api_file_v2") // set captured table                .username("root")                .password("root")                .startupOptions(StartupOptions.timestamp(1667232000000l))                .debeziumProperties(getDebeziumProperties())                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String                .build();        env                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")                .setParallelism(4)                .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering        env.execute();    }    private static Properties getDebeziumProperties(){        Properties properties = new Properties();        properties.setProperty("converters", "dateConverters");        //根据类在那个包下面修改        properties.setProperty("dateConverters.type", "com.belle.dc.stram.MySqlDateTimeConverter");        properties.setProperty("dateConverters.format.date", "yyyy-MM-dd");        properties.setProperty("dateConverters.format.time", "HH:mm:ss");        properties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");        properties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");        properties.setProperty("dateConverters.format.timestamp.zone", "UTC+8");        properties.setProperty("debezium.snapshot.locking.mode","none"); //全局读写锁，可能会影响在线业务，跳过锁设置        properties.setProperty("include.schema.changes", "true");        properties.setProperty("bigint.unsigned.handling.mode","long");        properties.setProperty("decimal.handling.mode","double");        return properties;    }}

1.3加入后采集的结果

{"before":null,"after":{"file_path":"magic-api/datasource/","file_content":"this is directory"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667445884000,"snapshot":"false","db":"magic_api","sequence":null,"table":"magic_api_file_v2","server_id":1,"gtid":null,"file":"","pos":1645,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791531,"transaction":null}{"before":null,"after":{"file_path":"magic-api/function/","file_content":"this is directory"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667445884000,"snapshot":"false","db":"magic_api","sequence":null,"table":"magic_api_file_v2","server_id":1,"gtid":null,"file":"","pos":1988,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791532,"transaction":null}{"before":null,"after":{"file_path":"magic-api/api/","file_content":"this is directory"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667445884000,"snapshot":"false","db":"magic_api","sequence":null,"table":"magic_api_file_v2","server_id":1,"gtid":null,"file":"","pos":2329,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791533,"transaction":null}{"before":null,"after":{"file_path":"magic-api/api/magic_api_test/","file_content":"this is directory"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667463052000,"snapshot":"false","db":"magic_api","sequence":null,"table":"magic_api_file_v2","server_id":1,"gtid":null,"file":"","pos":2665,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791533,"transaction":null}{"before":null,"after":{"file_path":"magic-api/api/magic_api_test/group.json","file_content":"{\r\n  \"properties\" : { },\r\n  \"id\" : \"57c3acdd91c34cd791eff5c187906231\",\r\n  \"name\" : \"magic_api_test\",\r\n  \"type\" : \"api\",\r\n  \"parentId\" : \"0\",\r\n  \"path\" : \"magic_api_test\",\r\n  \"createTime\" : 1667463052248,\r\n  \"updateTime\" : null,\r\n  \"createBy\" : null,\r\n  \"updateBy\" : null,\r\n  \"paths\" : [ ],\r\n  \"options\" : [ ]\r\n}"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667463052000,"snapshot":"false","db":"magic_api","sequence":null,"table":"magic_api_file_v2","server_id":1,"gtid":null,"file":"","pos":3016,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791534,"transaction":null}{"before":null,"after":{"file_path":"magic-api/api/magic_api_test/magic_api测试.ms","file_content":"{\r\n  \"properties\" : { },\r\n  \"id\" : \"2849b081c7bb4b0eb752c37020b77099\",\r\n  \"script\" : null,\r\n  \"groupId\" : \"57c3acdd91c34cd791eff5c187906231\",\r\n  \"name\" : \"magic_api测试\",\r\n  \"createTime\" : 1667463229393,\r\n  \"updateTime\" : null,\r\n  \"lock\" : null,\r\n  \"createBy\" : null,\r\n  \"updateBy\" : null,\r\n  \"path\" : \"/test\",\r\n  \"method\" : \"GET\",\r\n  \"parameters\" : [ ],\r\n  \"options\" : [ ],\r\n  \"requestBody\" : null,\r\n  \"headers\" : [ ],\r\n  \"paths\" : [ ],\r\n  \"responseBody\" : null,\r\n  \"description\" : null,\r\n  \"requestBodyDefinition\" : null,\r\n  \"responseBodyDefinition\" : null\r\n}\r\n================================\r\nreturn 'Hello magic-api'"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667463229000,"snapshot":"false","db":"magic_api","sequence":null,"table":"magic_api_file_v2","server_id":1,"gtid":null,"file":"","pos":4374,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791535,"transaction":null}{"before":{"file_path":"magic-api/api/magic_api_test/magic_api测试.ms","file_content":"{\r\n  \"properties\" : { },\r\n  \"id\" : \"2849b081c7bb4b0eb752c37020b77099\",\r\n  \"script\" : null,\r\n  \"groupId\" : \"57c3acdd91c34cd791eff5c187906231\",\r\n  \"name\" : \"magic_api测试\",\r\n  \"createTime\" : 1667463229393,\r\n  \"updateTime\" : null,\r\n  \"lock\" : null,\r\n  \"createBy\" : null,\r\n  \"updateBy\" : null,\r\n  \"path\" : \"/test\",\r\n  \"method\" : \"GET\",\r\n  \"parameters\" : [ ],\r\n  \"options\" : [ ],\r\n  \"requestBody\" : null,\r\n  \"headers\" : [ ],\r\n  \"paths\" : [ ],\r\n  \"responseBody\" : null,\r\n  \"description\" : null,\r\n  \"requestBodyDefinition\" : null,\r\n  \"responseBodyDefinition\" : null\r\n}\r\n================================\r\nreturn 'Hello magic-api'"},"after":{"file_path":"magic-api/api/magic_api_test/magic_api测试.ms","file_content":"{\r\n  \"properties\" : { },\r\n  \"id\" : \"2849b081c7bb4b0eb752c37020b77099\",\r\n  \"script\" : null,\r\n  \"groupId\" : \"57c3acdd91c34cd791eff5c187906231\",\r\n  \"name\" : \"magic_api测试\",\r\n  \"createTime\" : null,\r\n  \"updateTime\" : 1667463464144,\r\n  \"lock\" : null,\r\n  \"createBy\" : null,\r\n  \"updateBy\" : null,\r\n  \"path\" : \"/test\",\r\n  \"method\" : \"GET\",\r\n  \"parameters\" : [ ],\r\n  \"options\" : [ ],\r\n  \"requestBody\" : null,\r\n  \"headers\" : [ ],\r\n  \"paths\" : [ ],\r\n  \"responseBody\" : \"{\\n    \\\"code\\\": 1,\\n    \\\"message\\\": \\\"success\\\",\\n    \\\"data\\\": \\\"Hello magic-api\\\",\\n    \\\"timestamp\\\": 1667463381304,\\n    \\\"executeTime\\\": 2\\n}\",\r\n  \"description\" : null,\r\n  \"requestBodyDefinition\" : null,\r\n  \"responseBodyDefinition\" : {\r\n    \"name\" : \"\",\r\n    \"value\" : \"\",\r\n    \"description\" : \"\",\r\n    \"required\" : false,\r\n    \"dataType\" : \"Object\",\r\n    \"type\" : null,\r\n    \"defaultValue\" : null,\r\n    \"validateType\" : \"\",\r\n    \"error\" : \"\",\r\n    \"expression\" : \"\",\r\n    \"children\" : [ {\r\n      \"name\" : \"code\",\r\n      \"value\" : \"1\",\r\n      \"description\" : \"\",\r\n      \"required\" : false,\r\n      \"dataType\" : \"Integer\",\r\n      \"type\" : null,\r\n      \"defaultValue\" : null,\r\n      \"validateType\" : \"\",\r\n      \"error\" : \"\",\r\n      \"expression\" : \"\",\r\n      \"children\" : [ ]\r\n    }, {\r\n      \"name\" : \"message\",\r\n      \"value\" : \"success\",\r\n      \"description\" : \"\",\r\n      \"required\" : false,\r\n      \"dataType\" : \"String\",\r\n      \"type\" : null,\r\n      \"defaultValue\" : null,\r\n      \"validateType\" : \"\",\r\n      \"error\" : \"\",\r\n      \"expression\" : \"\",\r\n      \"children\" : [ ]\r\n    }, {\r\n      \"name\" : \"data\",\r\n      \"value\" : \"Hello magic-api\",\r\n      \"description\" : \"\",\r\n      \"required\" : false,\r\n      \"dataType\" : \"String\",\r\n      \"type\" : null,\r\n      \"defaultValue\" : null,\r\n      \"validateType\" : \"\",\r\n      \"error\" : \"\",\r\n      \"expression\" : \"\",\r\n      \"children\" : [ ]\r\n    }, {\r\n      \"name\" : \"timestamp\",\r\n      \"value\" : \"1667463381304\",\r\n      \"description\" : \"\",\r\n      \"required\" : false,\r\n      \"dataType\" : \"Long\",\r\n      \"type\" : null,\r\n      \"defaultValue\" : null,\r\n      \"validateType\" : \"\",\r\n      \"error\" : \"\",\r\n      \"expression\" : \"\",\r\n      \"children\" : [ ]\r\n    }, {\r\n      \"name\" : \"executeTime\",\r\n      \"value\" : \"2\",\r\n      \"description\" : \"\",\r\n      \"required\" : false,\r\n      \"dataType\" : \"Integer\",\r\n      \"type\" : null,\r\n      \"defaultValue\" : null,\r\n      \"validateType\" : \"\",\r\n      \"error\" : \"\",\r\n      \"expression\" : \"\",\r\n      \"children\" : [ ]\r\n    } ]\r\n  }\r\n}\r\n================================\r\nvar sql = \"\"\"\r\n    select * from magic_backup_record_v2\r\n\"\"\"\r\n\r\nreturn db.select(sql)"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667463464000,"snapshot":"false","db":"magic_api","sequence":null,"table":"magic_api_file_v2","server_id":1,"gtid":null,"file":"","pos":6330,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1668157791537,"transaction":null}{"before":null,"after":{"id":"1","claim_amount":0.0,"follow_result_type":"2","logistics_settle_flag":null,"submit_operater":"s","apply_operater":"s","submit_time":"2022-10-14 09:53:24","apply_time":"2022-10-14 09:53:30","claim_remark":"自动提报调度：异常售后自动提报为(实物追回)。","exception_type":"AG_RETURN_LOSS","apply_id":"df1bbf781000472b8988240a55f9fb73","apply_no":"1","order_sub_no":"1","out_order_id":"1","follow_status":"PROCESSED","follow_source":"oms","loss_reason":null,"responsibility_ascription":null,"trading_no":null,"dept_code":null,"update_time":"2022-10-14 09:55:41","create_time":"2022-10-14 09:55:37","total_ex_goods_num":0},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667546262000,"snapshot":"false","db":"flink_cdc_timestamp","sequence":null,"table":"table_flink_cdc_test1_default_problem","server_id":1,"gtid":null,"file":"","pos":16892,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791551,"transaction":null}{"before":null,"after":{"id":"2","claim_amount":0.0,"follow_result_type":"2","logistics_settle_flag":null,"submit_operater":"s","apply_operater":"s","submit_time":"2022-10-14 09:53:24","apply_time":"2022-10-14 09:53:30","claim_remark":"自动提报调度：异常售后自动提报为(实物追回)。","exception_type":"AG_RETURN_LOSS","apply_id":"df1bbf781000472b8988240a55f9fb73","apply_no":"1","order_sub_no":"1","out_order_id":"1","follow_status":"PROCESSED","follow_source":"oms","loss_reason":null,"responsibility_ascription":"None","trading_no":null,"dept_code":null,"update_time":"2022-10-14 10:19:49","create_time":"2022-10-14 09:55:37","total_ex_goods_num":0},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667546262000,"snapshot":"false","db":"flink_cdc_timestamp","sequence":null,"table":"table_flink_cdc_test1_default_problem","server_id":1,"gtid":null,"file":"","pos":17269,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791551,"transaction":null}{"before":null,"after":{"id":"4","claim_amount":0.0,"follow_result_type":"2","logistics_settle_flag":null,"submit_operater":"s","apply_operater":"s","submit_time":"2022-10-14 09:53:24","apply_time":"2022-10-14 09:53:30","claim_remark":"自动提报调度：异常售后自动提报为(实物追回)。","exception_type":"AG_RETURN_LOSS","apply_id":"df1bbf781000472b8988240a55f9fb73","apply_no":"1","order_sub_no":"1","out_order_id":"1","follow_status":"PROCESSED","follow_source":"oms","loss_reason":null,"responsibility_ascription":"None","trading_no":null,"dept_code":null,"update_time":"2022-10-14 10:19:49","create_time":"2022-10-14 09:55:37","total_ex_goods_num":0},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667546262000,"snapshot":"false","db":"flink_cdc_timestamp","sequence":null,"table":"table_flink_cdc_test1_default_problem","server_id":1,"gtid":null,"file":"","pos":17643,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1668157791552,"transaction":null}{"before":{"id":"4","claim_amount":0.0,"follow_result_type":"2","logistics_settle_flag":null,"submit_operater":"s","apply_operater":"s","submit_time":"2022-10-14 09:53:24","apply_time":"2022-10-14 09:53:30","claim_remark":"自动提报调度：异常售后自动提报为(实物追回)。","exception_type":"AG_RETURN_LOSS","apply_id":"df1bbf781000472b8988240a55f9fb73","apply_no":"1","order_sub_no":"1","out_order_id":"1","follow_status":"PROCESSED","follow_source":"oms","loss_reason":null,"responsibility_ascription":"None","trading_no":null,"dept_code":null,"update_time":"2022-10-14 10:19:49","create_time":"2022-10-14 09:55:37","total_ex_goods_num":0},"after":{"id":"4","claim_amount":0.0,"follow_result_type":"2","logistics_settle_flag":null,"submit_operater":"s","apply_operater":"s","submit_time":null,"apply_time":"2022-10-14 09:53:30","claim_remark":"自动提报调度：异常售后自动提报为(实物追回)。","exception_type":"AG_RETURN_LOSS","apply_id":"df1bbf781000472b8988240a55f9fb73","apply_no":"1","order_sub_no":"1","out_order_id":"1","follow_status":"PROCESSED","follow_source":"oms","loss_reason":null,"responsibility_ascription":"None","trading_no":null,"dept_code":null,"update_time":"2022-11-11 16:21:40","create_time":"2022-10-14 09:55:37","total_ex_goods_num":0},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1668154900000,"snapshot":"false","db":"flink_cdc_timestamp","sequence":null,"table":"table_flink_cdc_test1_default_problem","server_id":1,"gtid":null,"file":"","pos":509,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1668157791573,"transaction":null}