Details
-
Improvement
-
Status: Closed
-
Minor
-
Resolution: Fixed
-
1.12.0, 1.12.1
Description
Canal flat message json format has an 'es' field extracted from mysql binlog which means the row data real change time in mysql. It expressed the event time naturally but was ignored during deserialization.
{ "data": [ { "id": "111", "name": "scooter", "description": "Big 2-wheel scooter", "weight": "5.18" } ], "database": "inventory", "es": 1589373560000, "id": 9, "isDdl": false, "mysqlType": { "id": "INTEGER", "name": "VARCHAR(255)", "description": "VARCHAR(512)", "weight": "FLOAT" }, "old": [ { "weight": "5.15" } ], "pkNames": [ "id" ], "sql": "", "sqlType": { "id": 4, "name": 12, "description": 12, "weight": 7 }, "table": "products", "ts": 1589373560798, "type": "UPDATE" }
org.apache.flink.formats.json.canal. CanalJsonDeserializationSchema
private static RowType createJsonRowType(DataType databaseSchema) { // Canal JSON contains other information, e.g. "ts", "sql", but we don't need them return (RowType) DataTypes.ROW( DataTypes.FIELD("data", DataTypes.ARRAY(databaseSchema)), DataTypes.FIELD("old", DataTypes.ARRAY(databaseSchema)), DataTypes.FIELD("type", DataTypes.STRING()), DataTypes.FIELD("database", DataTypes.STRING()), DataTypes.FIELD("table", DataTypes.STRING())) .getLogicalType(); }
Attachments
Attachments
Issue Links
- links to