从Kafka导入Vertica时无法识别Epoch时间戳的问题排查

问题详情 #

导入查询语句 #

COPY public.from_kafka 
   SOURCE KafkaSource(stream='example_data|0|-2, example_data|1|-2',
                      brokers='kafka01.example.com:9092',
                      duration=interval '10000 milliseconds') PARSER KafkaJSONParser()
   REJECTED DATA AS TABLE public.rejections;

Kafka消息格式 #

{"location_id":30277, "start_date":1667911800000}

Vertica表结构 #

CREATE TABLE public.from_kafka
(
    location_id int NOT NULL,
    start_date timestamp NOT NULL
);

CREATE PROJECTION public.from_kafka /*+createtype(L)*/ 
(
 location_id ENCODING RLE,
 start_date ENCODING GCDDELTA
)
AS
 SELECT from_kafka.location_id,
        from_kafka.start_date
 FROM public.from_kafka
 ORDER BY from_kafka.start_date,
          from_kafka.location_id
SEGMENTED BY hash(from_kafka.location_id, from_kafka.start_date) ALL NODES KSAFE 1;

错误现象 #

执行导入后无新数据写入，public.rejections表显示拒绝原因：

Missing or null value for column with NOT NULL constraint [start_date]
但被拒绝的数据明确包含start_date字段的有效值。

原因分析 #

Vertica的KafkaJSONParser()无法直接将JSON中的long类型毫秒级时间戳转换为timestamp类型，解析过程中无法识别该字段的有效值，进而触发NOT NULL约束报错。

解决方案 #

通过Java程序消费原Kafka主题的消息，将JSON中的start_date字段从long类型毫秒时间戳转换为符合Vertica timestamp格式的字符串（例如yyyy-MM-dd HH:mm:ss或ISO 8601格式），然后将转换后的JSON消息写入新的Kafka主题，最后使用KafkaJSONParser()从新主题导入数据到Vertica。

内容的提问来源于stack exchange，提问作者Dean Taler