Блог

Используя updateRecord, я пытаюсь преобразовать строку json в допустимый json:

Входные данные:

 {
  "user" : "{"id":"1","name":"TEST"}"
}
 

Текущий вывод:

 {
  "user":"{id=1, name=TEST}"
}
 

Ожидаемый результат:

 {
  "user": { 
      "id": "1", 
      "name": "TEST" 
  }
}
 

Как я могу это исправить? Использовать другой процессор или скрипт? Можете ли вы дать мне совет?

С уважением!

updateRecord:

Вы можете использовать JoltTransformJSON processsor ( библиотека преобразования Jolt предназначена для преобразования JSON в другой JSON) со спецификацией

 [
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "u": "=split('"', @(1,user))",
      "user": "=join('',@(1,u))"
    }
  },
  {
    "operation": "remove",
    "spec": {
      "u": ""
    }
  }
]
 

где " фрагменты удаляются с помощью split, а затем функция join используется для объединения остальной части строки. Лишний созданный вспомогательный атрибут u удаляется на последнем шаге.

Редактировать: в зависимости от вашего редактирования используйте следующие преобразования

 [
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "user": "=split('"', @(1,user))",
      "id": "@(1,user[3])",
      "name": "@(1,user[7])"
    }
  },
  {
    "operation": "remove",
    "spec": {
      "user": ""
    }
  },
  {
    "operation": "shift",
    "spec": {
      "@": "user"
    }
  }
]
 

Причина, по которой updateRecord работает таким образом, заключается в том, что он выводит схему записи из входных записей и определяет, что существует пользователь поля с типом String .

Таким образом, даже если вы попытаетесь использовать ScriptedTransformRecord вместо updateRecord и попытаетесь использовать Record.setValue(fieldName,convertStringToMap(previousValue)) , это преобразует карту в строку, и вы все равно получите поле user со строковым типом. (setValue() не изменяет тип уже существующих полей)

Обходным путем может быть обогащение записи добавлением нового поля, скажем, user_map с типом Map(строка, строка).

В качестве альтернативы, просто создайте новую запись с новой схемой, используя ScriptedTransformRecord, например, используя следующий сценарий groovy, где JsonSlurper используется для отмены экранирования Json

 import groovy.json.JsonSlurper
import org.apache.nifi.serialization.SimpleRecordSchema
import org.apache.nifi.serialization.record.MapRecord
import org.apache.nifi.serialization.record.Record

def slurper = new JsonSlurper()
Record result = new MapRecord(new SimpleRecordSchema([]),[:]) // empty record
record.toMap().each { k,v ->
    result.setValue(k,slurper.parseText(v as String)) // adding a field Map<String,String>
}
return result