Great question! Handling heterogeneous order data from multiple sources is such a common pain point in integration workflows, and building a flexible, maintainable mapping layer is exactly the way to go. Here's a practical, scalable approach I've used successfully for similar projects:

Start by creating a single, editable file that defines your unified order format. This will be the only file you need to modify if your target structure changes later. You can use JSON Schema (great for cross-language compatibility), Protobuf, or even a simple dataclass if you're working in a specific language.

Example JSON Schema (unified_order_schema.json):

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "title": "UnifiedOrder",
  "required": ["order_id", "customer_name", "total_amount", "order_date"],
  "properties": {
    "order_id": {"type": "string", "description": "Unique identifier for the order"},
    "customer_name": {"type": "string", "description": "Full name of the customer"},
    "total_amount": {"type": "number", "description": "Total monetary value of the order"},
    "order_date": {"type": "string", "format": "date-time", "description": "Timestamp when the order was placed"}
  }
}

This schema acts as your north star—all source data will be mapped to align with this structure.

Instead of hardcoding mappings in your code, create separate configuration files (YAML works great here) for each data source. These files define how source fields map to your unified schema, plus any transformation rules needed (like date formatting or type conversions).

Example for Company A's CSV (company_a_mapping.yaml):

source_type: "csv"
file_pattern: "./data/company_a/*.csv"
field_mappings:
  order_id: "OrderNumber"  # Exact column name in Company A's CSV
  customer_name: "CustomerFullName"
  total_amount:
    source_field: "TotalValue"
    transform: "convert_to_float"
  order_date:
    source_field: "OrderTimestamp"
    transform: "parse_datetime('%Y-%m-%d %H:%M:%S')"

Example for Company B's TXT (pipe-separated) (company_b_mapping.yaml):

source_type: "txt"
file_pattern: "./data/company_b/*.txt"
delimiter: "|"
field_mappings:
  order_id: "ORD_ID"
  customer_name: "CUST_NAME"
  total_amount:
    source_field: "AMOUNT"
    transform: "convert_to_float"
  order_date:
    source_field: "DATE_PLACED"
    transform: "parse_datetime('%d/%m/%Y')"

The best part? Adding a new source or updating an existing mapping only requires editing these YAML files—no code changes needed.

Write a reusable script/service that reads these configuration files, loads the source data, applies the mapping rules, and outputs data that matches your unified schema. Here's a simplified Python example:

import csv
import yaml
import json
from datetime import datetime
from jsonschema import validate, ValidationError

def load_config(config_path):
    with open(config_path, "r") as f:
        return yaml.safe_load(f)

def load_unified_schema(schema_path):
    with open(schema_path, "r") as f:
        return json.load(f)

def transform_value(value, transform_rule):
    if not transform_rule:
        return value.strip()
    
    if transform_rule == "convert_to_float":
        return float(value.strip())
    
    if transform_rule.startswith("parse_datetime"):
        date_format = transform_rule.split("'")[1]
        return datetime.strptime(value.strip(), date_format).isoformat()
    
    # Add more transform functions (e.g., string cleanup, enum mapping) as needed
    return value.strip()

def process_source(source_config, unified_schema):
    unified_orders = []
    source_type = source_config["source_type"]
    
    # Handle CSV files
    if source_type == "csv":
        # Replace * with a sample filename for demo; in production, use glob to find all files
        with open(source_config["file_pattern"].replace("*", "sample_order"), "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                order = {}
                for target_field, mapping in source_config["field_mappings"].items():
                    if isinstance(mapping, str):
                        order[target_field] = row[mapping].strip()
                    else:
                        source_val = row[mapping["source_field"]]
                        order[target_field] = transform_value(source_val, mapping["transform"])
                
                # Validate against unified schema to catch errors early
                try:
                    validate(instance=order, schema=unified_schema)
                    unified_orders.append(order)
                except ValidationError as e:
                    print(f"Skipping invalid order (source: Company A): {row} | Error: {str(e)}")
                    # Log errors to a file or monitoring system in production
    
    # Add handlers for TXT, JSON, XML, etc., following the same pattern
    elif source_type == "txt":
        # Implement pipe-separated TXT processing here
        pass
    
    return unified_orders

# Usage example
unified_schema = load_unified_schema("./unified_order_schema.json")
company_a_config = load_config("./company_a_mapping.yaml")
company_a_unified_orders = process_source(company_a_config, unified_schema)

# Output to a unified JSON file (or send to a database/API)
with open("./unified_orders.json", "w") as f:
    json.dump(company_a_unified_orders, f, indent=2)

As shown in the code above, validating each transformed order against your unified schema is critical for reducing errors. You can also add:

• Logging for invalid records (with context about the source and error)
• Alerting for high error rates (e.g., if 10% of Company B's orders fail validation)
• Fallback logic for optional fields (e.g., default to null if a source field is missing)

Store your unified schema and mapping configs in a version control system (like Git) so your team can collaborate on changes. If you have non-technical stakeholders who need to edit mappings, you could even build a simple web UI that writes to these config files—though for most teams, YAML + Git is more than sufficient.

This approach ensures you have a single, changeable source of truth for your target format, keeps mapping logic decoupled from your core code, and makes it trivial to onboard new data sources or adjust your unified structure as business needs change.

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