Hi Bruce, By looking at the stacktrace, the exception complains that CSVRecordReader could not read record correctly at PutDatabaseRecord. Probably you need to use JsonTreeReader at PutDatabaseRecord since you used JsonRecordSetWriter at UpdateRecord. The error looks as if PutDatabaseRecord tried to parse JSON as CSV.
CSV -> UpdateRecord -> JSON -> PutDatabaseRecord Thanks, Koji On Mon, Aug 28, 2017 at 1:43 AM, Bruce Lowther <[email protected]> wrote: > 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. >
