版本:flink-doris-connector-1.17 1.4.0
通过SchemaChange 数据流(JsonDebeziumSchemaSerializer)写入doris库,doris库中的表设计了datetime和date字段,在写入时,拦截了一下数据并解析成json,无论对应字段采用Long型的时间戳,还是java自带的new date等方式,都无法使doris中的datetime和date字段存入值,其他字段正常
你直接输出下,看上游数据是什么?
{“op”:“c”,“before”:null,“after”:{“DATE”:1699424445,“OP”:“c”,“CITY”:8,“PV”:8.0,“TIMESTAMP”:1699424445,“USERNAME”:“8”,“PVCHANGE”:8.0,“AFTERPV”:8.0,“TIME”:1699424445,BEFOREPV":0,“SITEID”:“8”},“source”:{“schema”:“C##DBZUSER”,“commit_scn”:“14375964”,“lcr_position”:null,“rs_id”:null,“txId”:“0a0009006a0d0000”,“redo_thread”:1,“version”:“1.9.7.Final”,“ssn”:0,“sequence”:null,“connector”:“oracle”,“name”:“oracle_logminer”,“ts_ms”:1699424445000,“snapshot”:“false”,“db”:“ORCLPDB”,“table”:“SITE_VISIT”,“scn”:“14375963”},“ts_ms”:1699424514085,“transaction”:null}
这里的DATE、TIMESTAMP、TIME都是JSONOBJECT LONG型,分别对应doris字段的date datetime和BIGINT,然而只有BIGINT能存入数据,其他俩字段是NULL
嗯,默认返回都是时间戳,你可以自己定制convert进行转换,参考https://github.com/apache/doris-flink-connector/blob/master/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/mysql/DateToStringConverter.java
默认从数据库拿到的是时间戳也可以,我可以通过flinkcdc的算子进行数据拦截转换,主要我现在是拿到数据后,不知道应该把时间戳转换成什么类型,才能被doris识别并且导入
转成时间的格式就行,比如 yyyy-MM-dd HH:mm:ss
原来是字符串日期格式就行,感谢