NiFi - CaptureChangeMySQL将json转换为[“col_name”：“col_value”]格式

Question

prolog

MySQL表名：ar_tmp有两列id int和name int

to do

我执行sql

insert into ar_tmp (id, name) values (1, 4);

和CaptureChangeMySQL捕获此CDC和流程内容

{
    "type":"insert",
    "timestamp":1550221517000,
    "binlog_filename":"mysql-bin.013920",
    "binlog_position":241518646,
    "database":"platform_data",
    "table_name":"ar_tmp",
    "table_id":2899035,
    "columns":[
        {
            "id":1,
            "name":"id",
            "column_type":4,
            "value":1
        },
        {
            "id":2,
            "name":"name",
            "column_type":4,
            "value":4
        },
        {
            "id":3,
            "value":4
        }
    ]
}

但我想要这种格式的结果

{
    "type":"insert",
    "timestamp":1550221517000,
    "binlog_filename":"mysql-bin.013920",
    "binlog_position":241518646,
    "database":"platform_data",
    "table_name":"ar_tmp",
    "table_id":2899035,
    "columns":[
        {
            "id":1,
            "name":4
        }
    ]
}

要么

{
    "id":1,
    "name":4
}

solution

它可以通过使用jsonPath函数进行硬编码来完成

但也许要“很难”这样做，因为每列使用相同的代码使处理器冗余（比如50列）。更糟糕的是，当列名更改时，这是危险的。

有任何想法吗？

Answer 1

JoltTransformJSON可以帮助你。

尝试一下here

您的演示输入json的Jolt规范：

[
  {
    "operation": "shift",
    "spec": {
      "columns": {
        "*": {
          "value": "columns.@(1,name)"
        }
      },
      "*": "&"
    }
  }
]

结果是：

{
  "type" : "insert",
  "timestamp" : 1550221517000,
  "binlog_filename" : "mysql-bin.013920",
  "binlog_position" : 241518646,
  "database" : "platform_data",
  "table_name" : "ar_tmp",
  "table_id" : 2899035,
  "columns" : {
    "id" : 1,
    "name" : 4
  }
}