0
0

flink cdc DataStream api 时区问题

flink cdc DataStream api 时区问题

以postgrsql 作为数据源时，Date和timesatmp等类型cdc同步读出来时，会发现一下几个问题：

  1. 时间，日期等类型的数据对应的会转化为Int，long等类型。
  3. 源表同步后，时间相差8小时。这是因为时区不同的缘故。

源表：
在这里插入图片描述
sink 表：
在这里插入图片描述
解决方案：在自定义序列化时进行处理。
java code

  2. package pg.cdc.ds;import com.alibaba.fastjson.JSONObject;import com.ververica.cdc.debezium.DebeziumDeserializationSchema;import io.debezium.data.Envelope;import org.apache.flink.api.common.typeinfo.BasicTypeInfo;import org.apache.flink.api.common.typeinfo.TypeInformation;import org.apache.flink.util.Collector;import org.apache.kafka.connect.data.Field;import org.apache.kafka.connect.data.Schema;import org.apache.kafka.connect.data.Struct;import org.apache.kafka.connect.source.SourceRecord;import java.text.SimpleDateFormat;import java.time.ZoneId;import java.util.Date;import java.util.List;
  4. public class CustomerDeserialization implements DebeziumDeserializationSchema<String>{
  6.     ZoneId serverTimeZone;
  7.     @Override
  8.     public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
  10.         //1.创建JSON对象用于存储最终数据
  11.         JSONObject result = new JSONObject();
  12.         Struct value =(Struct) sourceRecord.value();
  14.         //2.获取库名&表名
  15.         Struct sourceStruct = value.getStruct("source");
  16.         String database = sourceStruct.getString("db");
  17.         String schema = sourceStruct.getString("schema");
  18.         String tableName = sourceStruct.getString("table");
  20.         //3.获取"before"数据
  21.         Struct before = value.getStruct("before");
  22.         JSONObject beforeJson = new JSONObject();
  23.         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  24.         SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd");if(before != null){
  25.             Schema beforeSchema = before.schema();
  26.             List<Field> beforeFields = beforeSchema.fields();for(Field field : beforeFields){
  27.                 Object beforeValue = before.get(field);if("int64".equals(field.schema().type().getName())&&"io.debezium.time.MicroTimestamp".equals(field.schema().name())){if(beforeValue != null){
  28.                         long times=(long) beforeValue / 1000;
  29.                         String dateTime = sdf.format(new Date((times -8*60*60*1000)));
  31.                         beforeJson.put(field.name(), dateTime);}}elseif("int64".equals(field.schema().type().getName())&&"io.debezium.time.NanoTimestamp".equals(field.schema().name())){if(beforeValue != null){
  32.                         long times=(long) beforeValue;
  33.                         String dateTime = sdf.format(new Date((times -8*60*60*1000)));
  34.                         beforeJson.put(field.name(), dateTime);}}elseif("int64".equals(field.schema().type().getName())&&"io.debezium.time.Timestamp".equals(field.schema().name())){if(beforeValue != null){
  35.                         long times=(long) beforeValue;
  36.                         String dateTime = sdf.format(new Date((times -8*60*60)));
  37.                         beforeJson.put(field.name(), dateTime);}}else if("int32".equals(field.schema().type().getName())&&"io.debezium.time.Date".equals(field.schema().name())){
  38.                     if(beforeValue != null){
  39.                         int times=(int) beforeValue;
  40.                         String dateTime = sdf1.format(new Date(times * 24 * 60 * 60L * 1000));
  41.                         beforeJson.put(field.name(), dateTime);}}else{
  42.                     beforeJson.put(field.name(), beforeValue);}}}
  43.         //4.获取"after"数据
  44.         Struct after = value.getStruct("after");
  45.         JSONObject afterJson = new JSONObject();if(after != null){
  46.             Schema afterSchema = after.schema();
  47.             List<Field> afterFields = afterSchema.fields();for(Field field : afterFields){
  48.                 Object afterValue = after.get(field);if("int64".equals(field.schema().type().getName())&&"io.debezium.time.MicroTimestamp".equals(field.schema().name())){if(afterValue != null){
  49.                         long times=(long) afterValue / 1000;
  50.                         String dateTime = sdf.format(new Date((times -8*60*60*1000)));
  52.                         afterJson.put(field.name(), dateTime);}}elseif("int64".equals(field.schema().type().getName())&&"io.debezium.time.NanoTimestamp".equals(field.schema().name())){if(afterValue != null){
  53.                         long times=(long) afterValue;
  54.                         String dateTime = sdf.format(new Date((times -8*60*60*1000)));
  55.                         afterJson.put(field.name(), dateTime);}}elseif("int64".equals(field.schema().type().getName())&&"io.debezium.time.Timestamp".equals(field.schema().name())){if(afterValue != null){
  56.                         long times=(long) afterValue;
  57.                         String dateTime = sdf.format(new Date((times -8*60*60)));
  58.                         afterJson.put(field.name(), dateTime);}}else if("int32".equals(field.schema().type().getName())&&"io.debezium.time.Date".equals(field.schema().name())){
  59.                     if(afterValue != null){
  60.                         int times=(int) afterValue;
  61.                         String dateTime = sdf1.format(new Date(times * 24 * 60 * 60L * 1000));
  62.                         afterJson.put(field.name(), dateTime);}}else{
  63.                     afterJson.put(field.name(), afterValue);}}}
  65.         //5.获取操作类型  CREATE UPDATE DELETE
  66.         Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  67.         String type= operation.toString().toLowerCase();if("create".equals(type)||"read".equals(type)){type="insert";}
  69.         //6.将字段写入JSON对象
  70.         result.put("database", database);
  71.         result.put("schema", schema);
  72.         result.put("tableName", tableName);
  73.         result.put("before", beforeJson);
  74.         result.put("after", afterJson);
  75.         result.put("type", type);
  76.         //7.输出数据
  77.         collector.collect(result.toJSONString());}
  79.     @Override
  80.     public TypeInformation<String>getProducedType(){return BasicTypeInfo.STRING_TYPE_INFO;}}

scala code

  1. import com.ververica.cdc.debezium.DebeziumDeserializationSchema
  2. import com.ververica.cdc.debezium.utils.TemporalConversions
  3. import io.debezium.time._
  4. import org.apache.flink.api.common.typeinfo.TypeInformation
  5. import org.apache.flink.types.Row
  6. import org.apache.flink.util.Collector
  7. import org.apache.kafka.connect.data.{SchemaBuilder, Struct}import org.apache.kafka.connect.source.SourceRecord
  8. import java.sql
  9. import java.time.{Instant, LocalDateTime, ZoneId}import scala.collection.JavaConverters._
  10. import scala.util.parsing.json.JSONObject
  11.  
  12.  
  13.  
  14. class StructDebeziumDeserializationSchema(serverTimeZone: String) extends DebeziumDeserializationSchema[Row]{
  15.  
  16.   override def deserialize(sourceRecord: SourceRecord, collector: Collector[Row]): Unit ={
  17.     // 解析主键
  18.     val key = sourceRecord.key().asInstanceOf[Struct]
  19.     val keyJs = parseStruct(key)
  20.  
  21.     // 解析值
  22.     val value = sourceRecord.value().asInstanceOf[Struct]
  23.     val source= value.getStruct("source")
  24.     val before = parseStruct(value.getStruct("before"))
  25.     val after = parseStruct(value.getStruct("after"))
  26.  
  27.     val row = Row.withNames()
  28.     row.setField("table", s"${source.get("db")}.${source.get("table")}")
  29.     row.setField("key", keyJs)
  30.     row.setField("op", value.get("op"))
  31.     row.setField("op_ts", LocalDateTime.ofInstant(Instant.ofEpochMilli(source.getInt64("ts_ms")), ZoneId.of(serverTimeZone)))
  32.     row.setField("current_ts", LocalDateTime.ofInstant(Instant.ofEpochMilli(value.getInt64("ts_ms")), ZoneId.of(serverTimeZone)))
  33.     row.setField("before", before)
  34.     row.setField("after", after)
  35.     collector.collect(row)}
  36.  
  37.   /** 解析[[Struct]]结构为json字符串 */
  38.   private def parseStruct(struct: Struct): String ={if(struct == null)return null
  39.     val map = struct.schema().fields().asScala.map(field =>{
  40.       val v= struct.get(field)
  41.       val typ = field.schema().name()
  42.       println(s"$v, $typ, ${field.name()}")
  43.       val value =v match {case long if long.isInstanceOf[Long]=> convertLongToTime(long.asInstanceOf[Long], typ)case iv if iv.isInstanceOf[Int]=> convertIntToDate(iv.asInstanceOf[Int], typ)case iv if iv == null => null
  44.         case _ => convertObjToTime(v, typ)}(field.name(), value)}).filter(_._2 != null).toMap
  45.     JSONObject.apply(map).toString()}
  46.  
  47.   /** 类型转换 */
  48.   private def convertObjToTime(obj: Any, typ: String): Any ={
  49.     typ match {case Time.SCHEMA_NAME | MicroTime.SCHEMA_NAME | NanoTime.SCHEMA_NAME =>
  50.         sql.Time.valueOf(TemporalConversions.toLocalTime(obj)).toString
  51.       case Timestamp.SCHEMA_NAME | MicroTimestamp.SCHEMA_NAME | NanoTimestamp.SCHEMA_NAME | ZonedTimestamp.SCHEMA_NAME =>
  52.         sql.Timestamp.valueOf(TemporalConversions.toLocalDateTime(obj, ZoneId.of(serverTimeZone))).toString
  53.       case _ => obj
  54.     }}
  55.  
  56.   /** long 转换为时间类型 */
  57.   private def convertLongToTime(obj: Long, typ: String): Any ={
  58.     val time_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Time")
  59.     val date_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Date")
  60.     val timestamp_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Timestamp")
  61.     typ match {case Time.SCHEMA_NAME =>
  62.         org.apache.kafka.connect.data.Time.toLogical(time_schema, obj.asInstanceOf[Int]).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalTime.toString
  63.       case MicroTime.SCHEMA_NAME =>
  64.         org.apache.kafka.connect.data.Time.toLogical(time_schema, (obj / 1000).asInstanceOf[Int]).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalTime.toString
  65.       case NanoTime.SCHEMA_NAME =>
  66.         org.apache.kafka.connect.data.Time.toLogical(time_schema, (obj / 1000 / 1000).asInstanceOf[Int]).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalTime.toString
  67.       case Timestamp.SCHEMA_NAME =>
  68.         val t = org.apache.kafka.connect.data.Timestamp.toLogical(timestamp_schema, obj).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalDateTime
  69.         java.sql.Timestamp.valueOf(t).toString
  70.       case MicroTimestamp.SCHEMA_NAME =>
  71.         val t = org.apache.kafka.connect.data.Timestamp.toLogical(timestamp_schema, obj / 1000).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalDateTime
  72.         java.sql.Timestamp.valueOf(t).toString
  73.       case NanoTimestamp.SCHEMA_NAME =>
  74.         val t = org.apache.kafka.connect.data.Timestamp.toLogical(timestamp_schema, obj / 1000 / 1000).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalDateTime
  75.         java.sql.Timestamp.valueOf(t).toString
  76.       case Date.SCHEMA_NAME =>
  77.         org.apache.kafka.connect.data.Date.toLogical(date_schema, obj.asInstanceOf[Int]).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalDate.toString
  78.       case _ => obj
  79.     }}
  80.  
  81.   private def convertIntToDate(obj:Int, typ: String): Any ={
  82.     val date_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Date")
  83.     typ match {case Date.SCHEMA_NAME =>
  84.         org.apache.kafka.connect.data.Date.toLogical(date_schema, obj).toInstant.atZone(ZoneId.of(serverTimeZone)).toLocalDate.toString
  85.       case _ => obj
  86.     }}
  87.  
  88.   override def getProducedType: TypeInformation[Row]={
  89.     TypeInformation.of(classOf[Row])}}

mysql cdc时区问题

mysql cdc也会出现上述时区问题，Debezium默认将MySQL中datetime类型转成UTC的时间戳({@link io.debezium.time.Timestamp})，时区是写死的无法更改，导致数据库中设置的UTC+8，到kafka中变成了多八个小时的long型时间戳 Debezium默认将MySQL中的timestamp类型转成UTC的字符串。

解决思路有两种：

1：自定义序列化方式的时候做时区转换。
2：自定义时间转换类，通过debezium配置文件指定转化格式。

这里主要使用第二种方式。

  1. package com.zmn.schema;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;
  3. /**
  4.  * 处理Debezium时间转换的问题
  5.  * Debezium默认将MySQL中datetime类型转成UTC的时间戳({@link io.debezium.time.Timestamp})，时区是写死的无法更改，
  6.  * 导致数据库中设置的UTC+8，到kafka中变成了多八个小时的long型时间戳
  7.  * Debezium默认将MySQL中的timestamp类型转成UTC的字符串。
  8.  * | mysql                               | mysql-binlog-connector                   | debezium                          |
  9.  * | ----------------------------------- | ---------------------------------------- | --------------------------------- |
  10.  * | date<br>(2021-01-28)| LocalDate<br/>(2021-01-28)| Integer<br/>(18655)|
  11.  * | time<br/>(17:29:04)| Duration<br/>(PT17H29M4S)| Long<br/>(62944000000)|
  12.  * | timestamp<br/>(2021-01-28 17:29:04)| ZonedDateTime<br/>(2021-01-28T09:29:04Z)| String<br/>(2021-01-28T09:29:04Z)|
  13.  * | Datetime<br/>(2021-01-28 17:29:04)| LocalDateTime<br/>(2021-01-28T17:29:04)| Long<br/>(1611854944000)|
  14.  *
  15.  * @see io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter
  16.  */
  17. public class MySqlDateTimeConverter implements CustomConverter<SchemaBuilder, RelationalColumn>{
  19.     private final static Logger logger = LoggerFactory.getLogger(MySqlDateTimeConverter.class);
  21.     private DateTimeFormatter dateFormatter = DateTimeFormatter.ISO_DATE;
  22.     private DateTimeFormatter timeFormatter = DateTimeFormatter.ISO_TIME;
  23.     private DateTimeFormatter datetimeFormatter = DateTimeFormatter.ISO_DATE_TIME;
  24.     private DateTimeFormatter timestampFormatter = DateTimeFormatter.ISO_DATE_TIME;
  26.     private ZoneId timestampZoneId = ZoneId.systemDefault();
  28.     @Override
  29.     public void configure(Properties props){
  30.         readProps(props, "format.date", p -> dateFormatter = DateTimeFormatter.ofPattern(p));
  31.         readProps(props, "format.time", p -> timeFormatter = DateTimeFormatter.ofPattern(p));
  32.         readProps(props, "format.datetime", p -> datetimeFormatter = DateTimeFormatter.ofPattern(p));
  33.         readProps(props, "format.timestamp", p -> timestampFormatter = DateTimeFormatter.ofPattern(p));
  34.         readProps(props, "format.timestamp.zone", z -> timestampZoneId = ZoneId.of(z));}
  36.     private void readProps(Properties properties, String settingKey, Consumer<String> callback){
  37.         String settingValue =(String) properties.get(settingKey);if(settingValue == null || settingValue.length()==0){return;}
  38.         try {
  39.             callback.accept(settingValue.trim());} catch (IllegalArgumentException | DateTimeException e){
  40.             logger.error("The {} setting is illegal: {}",settingKey,settingValue);
  41.             throw e;}}
  43.     @Override
  44.     public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBuilder> registration){
  45.         String sqlType = column.typeName().toUpperCase();
  46.         SchemaBuilder schemaBuilder = null;
  47.         Converter converter = null;if("DATE".equals(sqlType)){
  48.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.date.string");
  49.             converter = this::convertDate;}if("TIME".equals(sqlType)){
  50.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.time.string");
  51.             converter = this::convertTime;}if("DATETIME".equals(sqlType)){
  52.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.datetime.string");
  53.             converter = this::convertDateTime;}if("TIMESTAMP".equals(sqlType)){
  54.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.timestamp.string");
  55.             converter = this::convertTimestamp;}if(schemaBuilder != null){
  56.             registration.register(schemaBuilder, converter);}}
  58.     private String convertDate(Object input){if(input instanceof LocalDate){return dateFormatter.format((LocalDate) input);}if(input instanceof Integer){
  59.             LocalDate date= LocalDate.ofEpochDay((Integer) input);return dateFormatter.format(date);}return null;}
  61.     private String convertTime(Object input){if(input instanceof Duration){
  62.             Duration duration =(Duration) input;
  63.             long seconds = duration.getSeconds();
  64.             int nano= duration.getNano();
  65.             LocalTime time= LocalTime.ofSecondOfDay(seconds).withNano(nano);return timeFormatter.format(time);}return null;}
  67.     private String convertDateTime(Object input){if(input instanceof LocalDateTime){return datetimeFormatter.format((LocalDateTime) input);}return null;}
  69.     private String convertTimestamp(Object input){if(input instanceof ZonedDateTime){
  70.             // mysql的timestamp会转成UTC存储，这里的zonedDatetime都是UTC时间
  71.             ZonedDateTime zonedDateTime =(ZonedDateTime) input;
  72.             LocalDateTime localDateTime = zonedDateTime.withZoneSameInstant(timestampZoneId).toLocalDateTime();return timestampFormatter.format(localDateTime);}return null;}}
使用方式：
  2. StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();
  3.         Properties properties = new Properties();
  4.         properties.setProperty("snapshot.mode", "schema_only"); // 增量读取
  6.         //自定义时间转换配置
  7.         properties.setProperty("converters", "dateConverters");
  8.         properties.setProperty("dateConverters.type", "pg.cdc.ds.PgSQLDateTimeConverter");
  9.         properties.setProperty("dateConverters.format.date", "yyyy-MM-dd");
  10.         properties.setProperty("dateConverters.format.time", "HH:mm:ss");
  11.         properties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
  12.         properties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
  13.         properties.setProperty("dateConverters.format.timestamp.zone", "UTC+8");
  14.         properties.setProperty("debezium.snapshot.locking.mode","none"); //全局读写锁，可能会影响在线业务，跳过锁设置        
  15.         properties.setProperty("include.schema.changes", "true");
  16.         // 使用flink mysql cdc 发现bigint unsigned类型的字段，capture以后转成了字符串类型，
  17.        // 用的这个解析吧JsonDebeziumDeserializationSchema。
  18.         properties.setProperty("bigint.unsigned.handling.mode","long");
  19.         properties.setProperty("decimal.handling.mode","double");
  20.         
  21.         MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
  22.                 .hostname("192.168.10.102")
  23.                 .port(3306)
  24.                 .username("yusys")
  25.                 .password("yusys")
  26.                 .port(3306)
  27.                 .databaseList("gmall")
  28.                 .tableList("gmall.faker_user1")
  29.                 .deserializer(new JsonDebeziumDeserializationSchema())
  30.                 .debeziumProperties(properties)
  31.                 .serverId(5409)
  32.                 .build();
  33.                 
  34.                 
  35.       SingleOutputStreamOperator<string> dataSource =env
  36.                 .addSource(sourceFunction).setParallelism(10).name("binlog-source");
标签： flink java kafka
本文转载自: https://blog.csdn.net/hellowangxiansheng/article/details/130124580
版权归原作者 活在风浪里~ 所有， 如有侵权，请联系我们删除。
登录后发布评论

“flink cdc DataStream api 时区问题”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。