1.3.0
06/05/2017 12:31:48 EDT
Tagged nifi-1.3.0-RC1
I am building a workflow for receiving MQTT messages then insert into
database. Records transmitted by my simple IOT device arrive on MQTT queue
as a csv string and I'm trying to use record operators to insert into
database. My IOT device has no clock, so I am augmenting the received MQTT
messages by adding a Date time stamp property on the flow files using
UpdateAttribute immediately after receipt.
(I know I could use insert to database time.. but I want to try to get the
time stamp closest to the origination of data)
I am using AVRO schema registry and added property 'mqtt_house_sensor'
{
"type": "record",
"namespace": "bruceco",
"name": "mqttReading",
"fields": [
{ "name": "DID", "type": "string" },
{ "name": "Version", "type": "string" },
{ "name": "Temp", "type": "float" },
{ "name": "RHumid", "type": "float" },
{ "name": "ReceiveDT", "type": "string" }
]
}
sqlite event schema
sqlite> .schema mqtt_event
CREATE TABLE mqtt_event (id integer primary key autoincrement, dbdt
datetime default (datetime('now','localtime')), DID text, Version text,
Temp float, RHumid float, ReceiveDT datetime);
Using the PutDatabaseRecord processor* I can successfully *insert rows into
database with this schema.
However I cannot successfully update mqtt records with the datetime stamp
that I capture upon receive.
Example successful rows:
sqlite> select * from mqtt_event limit 2;
1|2017-08-27 11:45:09|v0.1|0x01|71.9599990844727|53.7000007629395|
2|2017-08-27 11:45:10|v0.1|0x01|72.6800003051758|54.0999984741211|
I am trying to use the use UpdateRecord processor to transfer date time
stamp property to record.
On the UpdateRecord processor, I set Replacement Value Strategy: Literal
Value; RecordReader to CSVReader and RecordWriter to JsonRecordSetWriter.
I have added a property name:
/ReceiveDT
And Value:
${ReceiveDT:format("yyyy/MM/dd HH:mm:ss")}
Actually I've tried several variations of the value but I can't seem to get
them to work.
Here is my current error result from the nifi-app.log:
2017-08-27 12:37:27,376 ERROR [Timer-Driven Process Thread-7]
o.a.n.p.standard.PutDatabaseRecord PutDatabaseRecord[id=015d1000-7fbf
-10d2-f081-23fe14b66269] Failed to process session due to
org.apache.nifi.processor.exception.ProcessException: Failed to process S
tandardFlowFileRecord[uuid=35dae543-f631-4843-9f8f-afc72f2fec77,claim=StandardContentClaim
[resourceClaim=StandardResourceClaim[id=
1503842607224-1, container=default, section=1], offset=168373,
length=122],offset=0,name=9285334964575,size=122] due to java.lang.R
untimeException: java.io.IOException: (line 2) invalid char between
encapsulated token and delimiter: {}
org.apache.nifi.processor.exception.ProcessException: Failed to process
StandardFlowFileRecord[uuid=35dae543-f631-4843-9f8f-afc72f2
fec77,claim=StandardContentClaim
[resourceClaim=StandardResourceClaim[id=1503842607224-1, container=default,
section=1], offset=168
373, length=122],offset=0,name=9285334964575,size=122] due to
java.lang.RuntimeException: java.io.IOException: (line 2) invalid cha
r between encapsulated token and delimiter
at
org.apache.nifi.processor.util.pattern.ExceptionHandler.lambda$createOnGroupError$14(ExceptionHandler.java:226)
at
org.apache.nifi.processor.util.pattern.ExceptionHandler.lambda$createOnError$13(ExceptionHandler.java:179)
at
org.apache.nifi.processor.util.pattern.ExceptionHandler$OnError.lambda$andThen$11(ExceptionHandler.java:54)
at
org.apache.nifi.processors.standard.PutDatabaseRecord.lambda$null$40(PutDatabaseRecord.java:404)
at
org.apache.nifi.processor.util.pattern.ExceptionHandler.execute(ExceptionHandler.java:148)
at
org.apache.nifi.processors.standard.PutDatabaseRecord.lambda$new$41(PutDatabaseRecord.java:357)
at
org.apache.nifi.processor.util.pattern.Put.putFlowFiles(Put.java:59)
at
org.apache.nifi.processor.util.pattern.Put.onTrigger(Put.java:101)
at
org.apache.nifi.processors.standard.PutDatabaseRecord.lambda$onTrigger$47(PutDatabaseRecord.java:689)
at
org.apache.nifi.processor.util.pattern.PartialFunctions.onTrigger(PartialFunctions.java:114)
at
org.apache.nifi.processor.util.pattern.RollbackOnFailure.onTrigger(RollbackOnFailure.java:184)
at
org.apache.nifi.processors.standard.PutDatabaseRecord.onTrigger(PutDatabaseRecord.java:689)
at
org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1120)
at
org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:147)
at
org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:47)
at
org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:132)
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:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.RuntimeException: java.io.IOException: (line 2)
invalid char between encapsulated token and delimiter
at
org.apache.commons.csv.CSVParser$1.getNextRecord(CSVParser.java:444)
at org.apache.commons.csv.CSVParser$1.hasNext(CSVParser.java:454)
at
org.apache.nifi.csv.CSVRecordReader.nextRecord(CSVRecordReader.java:73)
at
org.apache.nifi.processors.standard.PutDatabaseRecord.executeDML(PutDatabaseRecord.java:644)
at
org.apache.nifi.processors.standard.PutDatabaseRecord.lambda$null$39(PutDatabaseRecord.java:382)
at
org.apache.nifi.processor.util.pattern.ExceptionHandler.execute(ExceptionHandler.java:127)
... 18 common frames omitted
Caused by: java.io.IOException: (line 2) invalid char between encapsulated
token and delimiter
at
org.apache.commons.csv.Lexer.parseEncapsulatedToken(Lexer.java:275)
at org.apache.commons.csv.Lexer.nextToken(Lexer.java:152)
at org.apache.commons.csv.CSVParser.nextRecord(CSVParser.java:500)
at
org.apache.commons.csv.CSVParser$1.getNextRecord(CSVParser.java:441)
... 23 common frames omitted
2017-08-27 12:37:32,702 INFO [NiFi Web Server-905]
o.a.n.controller.StandardProcessorNode Stopping processor: class
org.apache.nifi.processors.standard.PutDatabaseRecord
Any suggestions on how to augment a record with an additional date time
stamp would be appreciate.
Thank you.
