如何使用Jolt过滤JSON数组，仅保留最新日期时间的条目

我正在处理一个JSON数组，其中每个对象可拼接出格式为YYYY-MM-DD HH:MM:SS的ScheduledDateUTC_Max字段。目标是通过Jolt转换该数组，仅保留该字段日期时间最新的对象。

输入JSON示例 #

[
  {
    "_id": {
      "$oid": "213asd4444-44asd44"
    },
    "PurchaseOrderNumber": "0085701091",
    "WorkflowId": "1243840",
    "CreatedDateUTC": "2023",
    "ExternalReference": "22261952",
    "Flag": "0",
    "NIT": "000890101676",
    "Tag03_ExternalReference": "early",
    "Warehouse_ExternalReference": "0085",
    "updateDate": {
      "$date": "2024-03-19T09:37:41.312Z"
    },
    "ScheduledDateUTC-Date": "2023-03-14",
    "ScheduledDateUTC-Hours": "07:30:00",
    "StatedTimestamp13UTC-Date": "2023-03-14",
    "StatedTimestamp13UTC-Hours": "05:29:35"
  },
  {
    "_id": {
      "$oid": "2134asdas444-44asda44"
    },
    "PurchaseOrderNumber": "0085701091",
    "WorkflowId": "1243839",
    "CreatedDateUTC": "2023",
    "ExternalReference": "22261952",
    "Flag": "0",
    "NIT": "000890101676",
    "Tag03_ExternalReference": "OnTime",
    "Warehouse_ExternalReference": "0085",
    "updateDate": {
      "$date": "2024-03-19T09:37:41.312Z"
    },
    "ScheduledDateUTC-Date": "2023-03-14",
    "ScheduledDateUTC-Hours": "06:20:00",
    "StatedTimestamp13UTC-Date": "2023-03-14",
    "StatedTimestamp13UTC-Hours": "05:29:35"
  },
  {
    "_id": {
      "$oid": "aasd2222sd-4444"
    },
    "PurchaseOrderNumber": "0085701091",
    "WorkflowId": "1243838",
    "CreatedDateUTC": "2023",
    "ExternalReference": "22261952",
    "Flag": "0",
    "NIT": "000890101676",
    "Tag03_ExternalReference": "OnTime",
    "Warehouse_ExternalReference": "0085",
    "updateDate": {
      "$date": "2024-03-19T09:37:41.312Z"
    },
    "ScheduledDateUTC-Date": "2023-03-14",
    "ScheduledDateUTC-Hours": "06:15:00",
    "StatedTimestamp13UTC-Date": "2023-03-14",
    "StatedTimestamp13UTC-Hours": "05:29:35"
  }
]

现有Jolt转换规则 #

[
  {
    "operation": "shift",
    "spec": {
      "*": {
        "ExternalReference": "[&1].ExternalReference",
        "Warehouse_ExternalReference": "[&1].Warehouse_ExternalReference",
        "PurchaseOrderNumber": "[&1].PurchaseOrderNumber",
        "CreatedDateUTC": "[&1].CreatedDateUTC",
        "NIT": "[&1].NIT",
        "ScheduledDateUTC-Date": "[&1].ScheduledDateUTC_Date",
        "ScheduledDateUTC-Hours": "[&1].ScheduledDateUTC_Hours",
        "StatedTimestamp13UTC-Date": "[&1].StatedTimestamp13UTC_Date",
        "StatedTimestamp13UTC-Hours": "[&1].StatedTimestamp13UTC_Hours"
      }
    }
 },
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "*": {
        "PurchaseOrderNumberSub": "=substring(@(1,PurchaseOrderNumber),4,10)",
        "CreatedDateUTC_PurchaseOrderNumber": "=concat(@(1,CreatedDateUTC),@(1,PurchaseOrderNumberSub))",
        "ScheduledDateUTC_Max": "=concat(@(1,ScheduledDateUTC_Date),' ',@(1,ScheduledDateUTC_Hours))"
      }
    }
 },
  {
    "operation": "remove",
    "spec": {
      "*": {
        "PurchaseOrderNumberSub": "",
        "CreatedDateUTC": ""
      }
    }
  },
  {
    "operation": "default",
    "spec": {
      "*": {
        "indicator": "${indicador}"
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      "*": {
        "ExternalReference": "[&1].ExternalReference",
        "Warehouse_ExternalReference": "[&1].Warehouse_ExternalReference",
        "CreatedDateUTC_PurchaseOrderNumber": "[&1].CreatedDateUTC_PurchaseOrderNumber",
        "NIT": "[&1].NIT",
        "ScheduledDateUTC_Date": "[&1].ScheduledDateUTC_Date",
        "ScheduledDateUTC_Hours": "[&1].ScheduledDateUTC_Hours",
        "StatedTimestamp13UTC_Date": "[&1].StatedTimestamp13UTC_Date",
        "StatedTimestamp13UTC_Hours": "[&1].StatedTimestamp13UTC_Hours",
        "indicator": "[&1].indicator",
        "ScheduledDateUTC_Max": "[&1].ScheduledDateUTC_Max"
      }
    }
  }
]

验证字段与期望输出 #

验证字段为："ScheduledDateUTC_Max": "2023-03-14 07:30:00"

期望输出：

[
 {
    "ExternalReference": "22261952",
    "Warehouse_ExternalReference": "0085",
    "CreatedDateUTC_PurchaseOrderNumber": "2023701091",
    "NIT": "000890101676",
    "ScheduledDateUTC_Date": "2023-03-14",
    "ScheduledDateUTC_Hours": "07:30:00",
    "StatedTimestamp13UTC_Date": "2023-03-14",
    "StatedTimestamp13UTC_Hours": "05:29:35",
    "indicator": "${indicador}"
    
 }
]

Jolt的modify-overwrite-beta操作支持部分数据处理，但不支持直接的日期时间比较，请问是否有办法通过Jolt实现该需求，或需要考虑其他工具/方案？

一、Jolt实现方案 #

Jolt本身没有直接的日期比较函数，但可以利用字符串排序特性（YYYY-MM-DD HH:MM:SS格式的字符串按字典序排序和时间顺序完全一致），通过以下步骤实现筛选：

完整Jolt规则 #

[
  // 第一步：完成原有字段处理（拼接、转换）
  {
    "operation": "shift",
    "spec": {
      "*": {
        "ExternalReference": "[&1].ExternalReference",
        "Warehouse_ExternalReference": "[&1].Warehouse_ExternalReference",
        "PurchaseOrderNumber": "[&1].PurchaseOrderNumber",
        "CreatedDateUTC": "[&1].CreatedDateUTC",
        "NIT": "[&1].NIT",
        "ScheduledDateUTC-Date": "[&1].ScheduledDateUTC_Date",
        "ScheduledDateUTC-Hours": "[&1].ScheduledDateUTC_Hours",
        "StatedTimestamp13UTC-Date": "[&1].StatedTimestamp13UTC_Date",
        "StatedTimestamp13UTC-Hours": "[&1].StatedTimestamp13UTC_Hours"
      }
    }
  },
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "*": {
        "PurchaseOrderNumberSub": "=substring(@(1,PurchaseOrderNumber),4,10)",
        "CreatedDateUTC_PurchaseOrderNumber": "=concat(@(1,CreatedDateUTC),@(1,PurchaseOrderNumberSub))",
        "ScheduledDateUTC_Max": "=concat(@(1,ScheduledDateUTC_Date),' ',@(1,ScheduledDateUTC_Hours))"
      }
    }
  },
  {
    "operation": "remove",
    "spec": {
      "*": {
        "PurchaseOrderNumberSub": "",
        "CreatedDateUTC": ""
      }
    }
  },
  {
    "operation": "default",
    "spec": {
      "*": {
        "indicator": "${indicador}"
      }
    }
  },
  // 第二步：将数组转为以ScheduledDateUTC_Max为键的对象，重复键会保留最后一个值（即最新时间）
  {
    "operation": "shift",
    "spec": {
      "*": {
        "@": "@(1,ScheduledDateUTC_Max)"
      }
    }
  },
  // 第三步：将对象转回数组，仅保留最新时间对应的对象
  {
    "operation": "shift",
    "spec": {
      "*": {
        "@": "[]"
      }
    }
  },
  // 第四步：移除ScheduledDateUTC_Max字段，匹配期望输出
  {
    "operation": "remove",
    "spec": {
      "*": {
        "ScheduledDateUTC_Max": ""
      }
    }
  }
]

原理说明 #

1. 先执行你原本的字段拼接、转换逻辑；
2. 通过shift将数组元素转为以时间字符串为键的对象，Jolt处理重复键时会自动保留最后一个值，而按字典序排序的时间字符串中，最新的时间会成为最后一个键；
3. 再将对象转回数组，此时数组仅包含最新时间对应的对象；
4. 最后移除临时的ScheduledDateUTC_Max字段，得到符合要求的输出。

二、替代工具方案 #

如果Jolt的间接实现方式不够灵活，可以考虑以下工具：

• jq：通过max_by函数直接按日期字段筛选，示例命令：

  jq 'max_by(.ScheduledDateUTC-Date + " " + .ScheduledDateUTC-Hours)' input.json
  

• Apache NiFi：使用ExecuteScript处理器编写Groovy/Java代码，直接解析日期并比较筛选；
• Python脚本：利用datetime模块解析时间字符串，遍历数组筛选出最大值对应的对象，实现逻辑更直观。

内容的提问来源于stack exchange，提问作者ODRAXIR