Flink SQL 数据sink到mysql时，非空列存在null值问题

flink sql 数据sink到mysql时，非空存在null值，插入mysql报错，配置下面的参数进行解决:

-- 默认值error
table.exec.sink.not-null-enforcer   

对表的NOT NULL列约束强制执行不能将空值插入到表中。

      Flink支持“error”（默认）和“drop”强制行为

      默认情况下，当将空值写入NOT NULL列时，Flink将检查值并引发运行时异常。

用户可以将行为更改为“drop”，以在不引发异常的情况下静默删除此类记录。

代码配置:

// instantiate table environment
val tEnv: TableEnvironment = ...


// access flink configuration
val configuration = tEnv.getConfig().getConfiguration()
// set low-level key-value options
configuration.setString("table.exec.sink.not-null-enforcer", "drop")

flink sql使用cdc时区差8小时问题

在连接参数中设置 ‘server-time-zone’ = ‘Asia/Shanghai’

比如：WITH (

‘connector’ = ‘mysql-cdc’,

‘hostname’ = ‘xxx’,

‘port’ = ‘3306’,

‘username’ = ‘root’,

‘password’ = ‘root’,

‘database-name’ = ‘xxx’,

‘table-name’ = ‘xxx’,

‘server-time-zone’ = ‘Asia/Shanghai’

不设置的话可能会改变MySQL中时间字段比如datetime减8小时

在sql语句中使用LOCALTIMESTAMP或者手动给时间戳加8小时而不要用current_date等

效果如图：

补充：

如果要sink到MySQL的话，在url后加&serverTimezone=Asia/Shanghai 否则时区也会对不上

或者在url上添加

jdbc:mysql://${hostname}/${db_name}?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&serverTimezone=Asia/Shanghai&useSSL=true&dontTrackOpenResources=true&defaultFetchSize=10000&useCursorFetch=true

‍