You have a few different kinds of transformations going on there: 1) Value -> Category, such as is_rain, season, and is_rushHour 2) Hoisting values from nested fields (possibly renaming the field), such as wind_speed 3) Fetch on match, such as PM10 and sensor_id, using nested values when a particular value is P1) In NiFi, some processors are better at different transformations than others. For the first kind, I was able to achieve this with EvaluateJsonPath followed by two UpdateAttribute processors: - EvaluateJsonPath extracts the "timestamp" field into an attribute called "dt". I couldn't use the actual "dt" field because it appears to be number of seconds since midnight or something, the date evaluates to 1/18/1970: - UpdateAttribute 1 extracts the hour and month values from the timestamp. Note the use of GMT timestamp in the functions, you may need to set that to something different (or exclude it altogether): - UpdateAttribute 2 performs the categorization logic, by checking if the month is between 12 and 2 (1 = winter), 3-5 (2 = spring), 6-8 (3=summer), 9-11 (4=fall), and also if the hour of day is between 8-9 AM or 5-6 PM for is_rushHour: The Expression for rush.hour is as follows: ${hour:ge(8):and(${hour:le(9)}):ifElse(1,${hour:ge(17):and(${hour:lt(19)}):ifElse(1,0)})} The Expression for season is as follows: ${month:gt(11):or(${month:lt(3)}):ifElse(1,
${month:ge(3):and(${month:lt(6)}):ifElse(2,
${month:ge(6):and(${month:lt(9)}):ifElse(3,4)})})} - JoltTransformJSON performs the other two types of transformation (with a "shift" spec), along with injecting the categorical fields using Expression Language (in a "default" spec), combined as a "Chain" spec: [
{
"operation": "shift",
"spec": {
"timestamp": "timestamp",
"wind": {
"deg": "wind_deg",
"speed": "wind_speed"
},
"main": {
"humidity": "humidity",
"pressure": "pressure",
"temp": "temperature"
},
"location": {
"longitude": "long",
"latitude": "lat"
},
"sensordatavalues": {
"*": {
"value_type": {
"P1": {
"@(2,value)": "PM10",
"@(2,id)": "sensor_id"
}
}
}
},
"cod": {
"200": {
"#1": "isRain"
},
"*": {
"#0": "isRain"
}
}
}
},
{
"operation": "default",
"spec": {
"season": "${season}",
"is_rushHour": "${rush.hour}"
}
}
] Note that I am assuming a "cod" value of 200 is rain and everything else is not. If there are other codes, you can add other blocks to the "cod" section of the spec after the 200 block. The * block handles anything not matched (basically an "else" clause). The flow is put together in the aforementioned order:
... View more
