CODEWITHSUNDEEP
arraysjsonapache-nifijolt
Thuy Le
Thuy Le

Reputation: 11

Nifi, Json data

I have JSON input data as

{
  "type": "insert",
  "timestamp": 1536959101000,
  "binlog_filename": "mysql-bin-changelog.234234",
  "binlog_position": 223,
  "database": "test",
  "table_name": "demo",
  "table_id": 138,
  "columns": [
    {
      "id": 1,
      "name": "id",
      "column_type": 12,
      "value": "IboECKV "
    },
    {
      "id": 2,
      "name": "col2",
      "column_type": 93,
      "value": "Fri Sep 14 21:05:02 UTC 2018"
    },
    {
      "id": 3,
      "name": "col3",
      "column_type": 4,
      "value": 10
    },
    {
      "id": 4,
      "name": "col4",
      "column_type": 4,
      "value": 0
    }
  ]
}

If column_type =93 (datetime): convert value to : yyyy-MM-dd HH:mm:ss.SSSZ

So the the target out put is

[
  {
    "id": "IboECKV "
  },
  {
    "col2": "2018-09-14 21:05:02.000Z"
  },
  {
    "col3": 10
  },
  {
    "col4": 0
  }
]

Do you know how to solved that case?

Many thanks,

Upvotes: 0

Views: 558

Answers (1)

Sivaprasanna Sethuraman
Sivaprasanna Sethuraman

Reputation: 4132

You can use ExecuteScript and leverage Groovy to do the parsing of Json input and the date and format it to however format you want using SimpleDateFormat.

A quick example:

import java.text.SimpleDateFormat 
import java.util.Date
import groovy.json.JsonSlurper
import groovy.json.JsonOutput
import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets

flowFile = session.get()
if(!flowFile)return
def text = ''

session.read(flowFile, {inputStream ->
  text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
} as InputStreamCallback)

def jsonSlurper = new JsonSlurper()
def object = jsonSlurper.parseText(text)

def columnsSize = object.columns.size

0.upto(columnsSize - 1) {
    if (object.columns[it].column_type == 93 ) {
        oldDate = object.columns[it].value
        sdfmt2= new SimpleDateFormat('dd-M-yyyy')
        parsedDate = sdfmt2.parse(oldDate)
        object.columns[it].value = parsedDate
        output = JsonOutput.toJson(object)

        flowFile = session.write(flowFile, {outputStream ->
            outputStream.write(output.getBytes(StandardCharsets.UTF_8))
        } as OutputStreamCallback)
        session.transfer(flowFile, REL_SUCCESS)
    }
}

Upvotes: 1

Related Questions