My goal is to read time-series data from CSV file and write the records to Kudu DB.
So the flow setup reads CSV file using FetchFile processor --> updates records values using UpdateRecord processor --> Writes values to Kudu using PutKudu processor.
The data in my CSV file looks as following:
Name,TS,Value,Error
AHU01_TT01_02_102,44195.00708,20.03832817,257
AHU01_TT01_02_102,44195.01417,20.01832008,257
AHU01_TT01_02_102,44195.02125,19.92832756,257
AHU01_TT01_02_102,44195.02833,19.93416214,257
The purpose of the UpdateRecord processor is to convert the epoch timestamp from Microsoft epoch format (seconds since 31/12/1899) to ISO Format date string or any other format acceptable by Kudu that can be read by any type of Record Reader.
So far the flow looks as follows:
The UpdateRecord 1.11.4 processor defined as follows:
Where it mostly used to convert the sample_time value from Microsoft timestamp format (seconds since 31/12/1899) to yyyy-MM-dd HH:mm:ssZ ISO-8601 format.
The CSV file samples reader (Record Reader) is defined as follows:
The Record Writer is defined as follows:
And finally the PutKudu Record Reader is defined as follows:
If the sample_time was not in as such complicated epoch format, I wouldn't need the UpdateRecord to convert the timestamp, as the last PutKude Records Reader service would do just that...
Anyhow, the flow returns the following failure, that I'm not sure how to handle it, and what's the problem with the output/input sample_time format:
If I understand correctly how the Record Reader/Writer schema works, it should process the sample_time in the following way: Get value from CSV as float format (using record reader) --> convert the value in any way using the UpdateRecord processor --> Write value in another format (string) using Record Writer...
How came this does not work?
Hopefully anybody can sort this out...
After playing quite a lot with different date formats and trying different approaches to convert the MS epoch to ISO-8601, I've finally came to a solution that worked for my use-case with a very small CSV, and here's what I've did:
First I've changed the UpdateRecord /sample_time to only convert from MS epoch to a standard UNIX epoch, leaving the output value as long integer:
Then in both the Record Writer and in the PutKudu Record Reader, I've set the sample_time as {"type": "long", "logicalType" : "timestamp-millis"}:
So far, it did the job and the values are written to Kudu.
But when I've then tried with the original 33Mb CSV file, containing all the samples, it failed again with a new error:
Not sure what does it mean...whither it's some row in the CSV that it's failing to process or something in the conversion that fails....
