基于运费规则的Jolt转换：寻求替代实现方案

需求说明 #

• 若输入JSON的任一订单元素中存在parentFacilityTypeId值为DISTRIBUTION_CENTER的发货项，则仅输出该订单的Order_ID和Shipment_ID。
• 若订单中无上述发货项，则需额外输出orderAdjustmentTypeId为SHIPPING_CHARGES对应的运费Amount，同时包含Order_ID和Shipment_ID。

输入JSON #

[
  {
    "orderId": "16487",
    "orderAdjustments": [
      {
        "amount": 0,
        "orderAdjustmentTypeId": "DONATION_ADJUSTMENT"
      },
      {
        "amount": 15.95,
        "orderAdjustmentTypeId": "SHIPPING_CHARGES"
      }
    ],
    "shipments": [
      {
        "shipmentId": "0001",
        "shipmentItems": [
          {
            "parentFacilityTypeId": "PHYSICAL_STORE",
            "quantity": 1
          },
          {
            "parentFacilityTypeId": "DISTRIBUTION_CENTER",
            "quantity": 1
          }
        ]
      }
    ]
  },
  {
    "orderId": "16488",
    "orderAdjustments": [
      {
        "amount": 10,
        "orderAdjustmentTypeId": "DONATION_ADJUSTMENT"
      },
      {
        "amount": 25.95,
        "orderAdjustmentTypeId": "SHIPPING_CHARGES"
      }
    ],
    "shipments": [
      {
        "shipmentId": "0001",
        "shipmentItems": [
          {
            "parentFacilityTypeId": "PHYSICAL_STORE",
            "quantity": 1
          }
        ]
      }
    ]
  },
  {
    "orderId": "16489",
    "orderAdjustments": [
      {
        "amount": 10,
        "orderAdjustmentTypeId": "DONATION_ADJUSTMENT"
      },
      {
        "amount": 25.95,
        "orderAdjustmentTypeId": "SHIPPING_CHARGES"
      }
    ],
    "shipments": [
      {
        "shipmentId": "0001",
        "shipmentItems": [
          {
            "parentFacilityTypeId": "DISTRIBUTION_CENTER",
            "quantity": 1
          }
        ]
      }
    ]
  }
]

预期输出JSON #

[
  {
    "Order_ID": "16487",
    "Shipment_ID": "0001"
  },
  {
    "Order_ID": "16488",
    "Amount": 25.95,
    "Shipment_ID": "0001"
  },
  {
    "Order_ID": "16489",
    "Shipment_ID": "0001"
  }
]

当前已实现的Jolt Spec #

[
  {
    "operation": "shift",
    "spec": {
      "*": {
        "shipments": {
          "*": {
            "shipmentItems": {
              "*": {
                "parentFacilityTypeId": {
                  "DISTRIBUTION_CENTER": {
                    "#Y": "[&7].fulfilledFromWH"
                  }
                }
              }
            }
          }
        },
        "@": "[&]"
      }
    }
  },
  {
    "operation": "modify-default-beta",
    "spec": {
      "*": {
        "fulfilledFromWH": "N"
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      "*": {
        "fulfilledFromWH": {
          "N": {
            "@(2,orderAdjustments)": {
              "*": {
                "orderAdjustmentTypeId": {
                  "SHIPPING_CHARGES": {
                    "@(2,amount)": "[&7].shippingAmount"
                  }
                }
              }
            }
          }
        },
        "@": "[&]"
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      "*": {
        "orderId": "[&1].Order_ID",
        "shippingAmount": "[&1].Amount",
        "shipments": {
          "*": {
            "shipmentId": "[&3].Shipment_ID"
          }
        }
      }
    }
  }
]

我已经通过上述Jolt Spec实现了需求并得到预期输出，现在想咨询是否有其他技术方案可以实现该需求，恳请提供帮助。

替代技术方案 #

方案1：Python脚本实现

通过遍历JSON结构直接进行条件判断和数据提取，代码简洁易读，适合小到中等数据量的处理。

import json

def process_orders(input_data):
    result = []
    for order in input_data:
        order_id = order["orderId"]
        shipment_id = order["shipments"][0]["shipmentId"]
        
        # 检查订单是否包含DISTRIBUTION_CENTER的发货项
        has_dist_center = any(
            item["parentFacilityTypeId"] == "DISTRIBUTION_CENTER"
            for shipment in order["shipments"]
            for item in shipment["shipmentItems"]
        )
        
        output_item = {
            "Order_ID": order_id,
            "Shipment_ID": shipment_id
        }
        
        # 无DISTRIBUTION_CENTER时添加运费金额
        if not has_dist_center:
            shipping_amount = next(
                adj["amount"] for adj in order["orderAdjustments"]
                if adj["orderAdjustmentTypeId"] == "SHIPPING_CHARGES"
            )
            output_item["Amount"] = shipping_amount
        
        result.append(output_item)
    return result

# 加载输入并处理
if __name__ == "__main__":
    with open("input.json", "r") as f:
        input_json = json.load(f)
    
    output_json = process_orders(input_json)
    print(json.dumps(output_json, indent=2))

方案2：Java Stream API实现

适合Java技术栈项目，利用流式处理和Jackson库解析JSON，可集成到后端服务中。

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class OrderShippingProcessor {
    public static void main(String[] args) throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        JsonNode inputArray = mapper.readTree(new File("input.json"));
        List<ObjectNode> resultList = new ArrayList<>();

        for (JsonNode orderNode : inputArray) {
            String orderId = orderNode.get("orderId").asText();
            JsonNode shipmentNode = orderNode.get("shipments").get(0);
            String shipmentId = shipmentNode.get("shipmentId").asText();
            
            // 判断是否存在DISTRIBUTION_CENTER发货项
            boolean hasDistCenter = false;
            Iterator<JsonNode> itemIterator = shipmentNode.get("shipmentItems").elements();
            while (itemIterator.hasNext()) {
                JsonNode item = itemIterator.next();
                if ("DISTRIBUTION_CENTER".equals(item.get("parentFacilityTypeId").asText())) {
                    hasDistCenter = true;
                    break;
                }
            }

            ObjectNode outputNode = mapper.createObjectNode();
            outputNode.put("Order_ID", orderId);
            outputNode.put("Shipment_ID", shipmentId);

            // 无DISTRIBUTION_CENTER时添加运费
            if (!hasDistCenter) {
                Iterator<JsonNode> adjIterator = orderNode.get("orderAdjustments").elements();
                while (adjIterator.hasNext()) {
                    JsonNode adj = adjIterator.next();
                    if ("SHIPPING_CHARGES".equals(adj.get("orderAdjustmentTypeId").asText())) {
                        outputNode.put("Amount", adj.get("amount").asDouble());
                        break;
                    }
                }
            }
            resultList.add(outputNode);
        }

        // 输出格式化后的结果
        System.out.println(mapper.writerWithDefaultPrettyPrinter().writeValueAsString(resultList));
    }
}

方案3：Apache Spark SQL实现

适合大数据量场景，利用Spark的分布式处理能力高效处理JSON数据。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object ShippingChargeProcessor {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("ShippingChargeProcessor")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    // 加载输入JSON为DataFrame
    val inputDF = spark.read.json("input.json")

    // 标记每个订单是否包含DISTRIBUTION_CENTER发货项
    val hasDistCenterDF = inputDF.select(
      $"orderId",
      explode($"shipments").alias("shipment"),
      explode($"shipments.shipmentItems").alias("item")
    ).groupBy($"orderId", $"shipment.shipmentId")
      .agg(max(when($"item.parentFacilityTypeId" === "DISTRIBUTION_CENTER", 1).otherwise(0)).alias("has_dist_center"))

    // 提取每个订单的运费金额
    val shippingAmountDF = inputDF.select(
      $"orderId",
      explode($"orderAdjustments").alias("adj")
    ).filter($"adj.orderAdjustmentTypeId" === "SHIPPING_CHARGES")
      .select($"orderId", $"adj.amount".alias("Amount"))

    // 关联数据生成最终结果
    val resultDF = hasDistCenterDF.join(shippingAmountDF, Seq("orderId"), "left_outer")
      .select(
        $"orderId".alias("Order_ID"),
        $"shipmentId".alias("Shipment_ID"),
        when($"has_dist_center" === 0, $"Amount").otherwise(null).alias("Amount")
      ).drop("has_dist_center")

    // 输出结果
    resultDF.show(false)
    resultDF.write.json("output.json")

    spark.stop()
  }
}

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