如何在NiFi ValidateRecord处理器/JsonRecordSetWriter中将时间戳序列化为Json字段

Question

如何将时间戳序列化为NiFi中的Json字段validateRecordprocessor/JSONRecordSetWriter。

在输入时，我有一个CSV文件，其中有一个时间戳列，格式为yyyy-mm-dd hh:mm:ss.sss。在我的NiFi流中，我有一个ValidateRecord处理器，它使用CSVReader进行读取，并使用JSONRecordSetWriter作为编写器。它们都使用Avro模式，时间戳字段定义为

"fields" : [ {
    "name" : "timestamp",
    "type" : {
            "type" : "long",
            "logicalType" : "timestamp-millis"
        },
    "doc" : "Type inferred from '2016/10/08 07:51:00.000'"
    }, {
    ...

当一个字段值为2016-10-08 07:51:0.000的记录出现时，NiFi日志中出现一个异常：

2018-10-18 17:05:59,135 ERROR [Timer-Driven Process Thread-1] o.a.n.processors.standard.ValidateRecord ValidateRecord[id=3d44915d-a52a-3eb0-1ae1-7b0cbe4b1a03] Failed to write MapRecord[{timestamp=2016-10-08 07:51:00.0, ...  ] with schema {"type":"record","name":"redfunnel","doc":"Schema generated by Kite","fields":[{"name":"timestamp","type":{"type":"long","logicalType":"timestamp-millis"},"doc":"Type inferred from '2016/10/08 07:51:00.000'"},{ .... }]} as a JSON Object due to java.lang.IllegalStateException: No ObjectCodec defined for the generator, can only serialize simple wrapper types (type passed java.sql.Timestamp): java.lang.IllegalStateException: No ObjectCodec defined for the generator, can only serialize simple wrapper types (type passed java.sql.Timestamp)
java.lang.IllegalStateException: No ObjectCodec defined for the generator, can only serialize simple wrapper types (type passed java.sql.Timestamp)
    at org.codehaus.jackson.impl.JsonGeneratorBase._writeSimpleObject(JsonGeneratorBase.java:556)
    at org.codehaus.jackson.impl.JsonGeneratorBase.writeObject(JsonGeneratorBase.java:317)
    at org.apache.nifi.json.WriteJsonResult.writeRawValue(WriteJsonResult.java:267)
    at org.apache.nifi.json.WriteJsonResult.writeRecord(WriteJsonResult.java:201)
    at org.apache.nifi.json.WriteJsonResult.writeRawRecord(WriteJsonResult.java:149)
    at org.apache.nifi.processors.standard.ValidateRecord.onTrigger(ValidateRecord.java:342)
    at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27)
    at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1165)
    at org.apache.nifi.controller.tasks.ConnectableTask.invoke(ConnectableTask.java:203)
    at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:117)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

在我的JSONRecordsetWriter的属性中，我尝试将时间戳的编写格式指定为yyyy-mm-dd hh:mm:ss.sss

但不幸的是没有成功，我仍然在NiFi日志中得到相同的异常。

这是否意味着JSONRecordSetWriter默认情况下不能序列化java.time.timestamp，即使它有Timestamp Format属性来进行配置呢？

是否可以使用开箱即用的NiFi组件根据自定义格式编写时间戳，或者我必须修改JSONRecordsetWriter？

Answer 1

似乎我找到了一个在我的情况下工作的配置。

我不得不将模式分成两部分：一个用于输入，另一个用于输出。

因此，schema1将时间戳字段定义为：

{
    "name" : "timestamp",
    "type" : "string",
    "doc" : "Type inferred from '2016/10/08 07:51:00.000'"
}

{
    "name" : "timestamp",
    "type" : {
        "type" : "long",
        "logicalType" : "timestamp-millis"
    },
    "doc" : "Type inferred from '2016/10/08 07:51:00.000'"
}

null

希望这能在类似的情况下对其他人有所帮助。