Koji, thank you for your feedback. With a little more review and brain calories, I realized that I had incorrectly accepted default option for 'Skip header line'.
Skip Header Line: false --translate--> Consider first line to be header. In my case this is not possible as my MQTT feed does not provide a header line, just single data row in tsv format. I think the UpdateAttribute processor was seeing a 'Temp' header value and trying to translate it as a float. Once I found that configuration issue, I am now getting a ReceiveDT stamp as soon as MQTT Message is received from source. Thank you for your suggestions. Spurred me on to try different alternatives Bruce W. Lowther On Mon, Aug 28, 2017 at 2:49 AM, Koji Kawamura <[email protected]> wrote: > 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. > > >
