如何编写不同表字段映射脚本

本文目录导读：

1. 基础字段映射脚本
2. 复杂字段映射（含转换函数）
3. 数据库表映射（SQL实现）
4. 自动化映射脚本
5. 批量映射处理
6. 最佳实践建议

我来详细解释如何编写不同表字段映射脚本,包含多种常见场景和实现方式。

基础字段映射脚本

JSON配置映射

# mapping_config.py
FIELD_MAPPING = {
    "source_table": {
        "user_id": "target_customer_id",
        "user_name": "customer_name",
        "user_email": "email_address",
        "user_phone": "phone_number",
        "created_at": "registration_date"
    }
}
# 映射执行脚本
def apply_mapping(source_data, mapping):
    """应用字段映射"""
    mapped_data = {}
    for source_field, target_field in mapping.items():
        if source_field in source_data:
            mapped_data[target_field] = source_data[source_field]
    return mapped_data
# 使用示例
source_record = {
    "user_id": 1001,
    "user_name": "张三",
    "user_email": "zhangsan@example.com"
}
mapped_record = apply_mapping(source_record, FIELD_MAPPING["source_table"])
print(mapped_record)
# 输出: {'target_customer_id': 1001, 'customer_name': '张三', 'email_address': 'zhangsan@example.com'}

复杂字段映射（含转换函数）

# complex_mapping.py
from datetime import datetime
import hashlib
import re
class FieldMapper:
    def __init__(self):
        self.mapping_rules = {
            "user_table": {
                "id": "customer_id",
                "name": "full_name",
                "email": "email",
                "phone": ("phone_number", self.clean_phone),
                "status": ("status_code", self.map_status),
                "birth_date": ("age", self.calculate_age),
                "password": ("password_hash", self.hash_password),
                "address": ("full_address", self.combine_address)
            }
        }
    def clean_phone(self, phone):
        """清理电话号码格式"""
        if not phone:
            return None
        # 移除所有非数字字符
        cleaned = re.sub(r'\D', '', phone)
        return f"+86{cleaned}" if len(cleaned) == 11 else cleaned
    def map_status(self, status):
        """状态映射"""
        status_map = {
            "active": "A",
            "inactive": "I",
            "pending": "P",
            "deleted": "D"
        }
        return status_map.get(status, "U")
    def calculate_age(self, birth_date):
        """计算年龄"""
        if not birth_date:
            return None
        birth = datetime.strptime(birth_date, "%Y-%m-%d")
        today = datetime.now()
        return today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day))
    def hash_password(self, password):
        """密码加密"""
        if not password:
            return None
        return hashlib.sha256(password.encode()).hexdigest()
    def combine_address(self, address_data):
        """组合地址"""
        if not address_data:
            return None
        parts = [
            address_data.get("province", ""),
            address_data.get("city", ""),
            address_data.get("district", ""),
            address_data.get("detail", "")
        ]
        return " ".join(filter(None, parts))
    def map_record(self, source_table, source_data):
        """执行映射"""
        if source_table not in self.mapping_rules:
            raise ValueError(f"Unknown source table: {source_table}")
        rules = self.mapping_rules[source_table]
        mapped_data = {}
        for source_field, target_rule in rules.items():
            if source_field not in source_data:
                continue
            if isinstance(target_rule, tuple):
                target_field, transform_func = target_rule
                mapped_data[target_field] = transform_func(source_data[source_field])
            else:
                mapped_data[target_rule] = source_data[source_field]
        return mapped_data
# 使用示例
mapper = FieldMapper()
source_data = {
    "id": 1001,
    "name": "张三",
    "email": "zhangsan@example.com",
    "phone": "13800138000",
    "status": "active",
    "birth_date": "1990-01-15",
    "password": "mypassword123",
    "address": {
        "province": "广东省",
        "city": "深圳市",
        "district": "南山区",
        "detail": "科技园路1号"
    }
}
result = mapper.map_record("user_table", source_data)
print(result)

数据库表映射（SQL实现）

-- 字段映射查询
CREATE OR REPLACE FUNCTION map_table_data()
RETURNS TABLE (
    target_id INTEGER,
    target_name VARCHAR,
    target_email VARCHAR,
    target_phone VARCHAR
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        s.user_id AS target_id,
        s.user_name AS target_name,
        COALESCE(s.email, s.contact_email) AS target_email,
        REGEXP_REPLACE(s.phone, '[^0-9]', '', 'g') AS target_phone
    FROM source_table s
    WHERE s.is_active = true;
END;
$$ LANGUAGE plpgsql;
-- 数据迁移脚本
INSERT INTO target_table (
    customer_id,
    full_name,
    email_address,
    phone_number,
    created_date
)
SELECT 
    s.id,
    s.name,
    CASE 
        WHEN s.email IS NOT NULL THEN s.email
        WHEN s.username LIKE '%@%' THEN s.username
        ELSE CONCAT(s.name, '@default.com')
    END AS email,
    s.phone,
    CURRENT_TIMESTAMP
FROM source_users s
WHERE NOT EXISTS (
    SELECT 1 FROM target_table t 
    WHERE t.customer_id = s.id
);

自动化映射脚本

# auto_mapper.py
import json
import yaml
from typing import Dict, Any, Callable, Optional
class AutoFieldMapper:
    def __init__(self, config_file: str = None):
        self.mappings = {}
        self.transformers = {}
        self.validators = {}
        if config_file:
            self.load_config(config_file)
    def load_config(self, config_file: str):
        """加载映射配置"""
        with open(config_file, 'r', encoding='utf-8') as f:
            if config_file.endswith('.json'):
                config = json.load(f)
            elif config_file.endswith(('.yaml', '.yml')):
                config = yaml.safe_load(f)
            else:
                raise ValueError("Unsupported config format")
        self.mappings = config.get('mappings', {})
    def register_transformer(self, name: str, func: Callable):
        """注册转换函数"""
        self.transformers[name] = func
    def register_validator(self, name: str, func: Callable):
        """注册验证器"""
        self.validators[name] = func
    def map_fields(self, source_data: Dict, mapping_name: str) -> Dict:
        """执行字段映射"""
        if mapping_name not in self.mappings:
            raise ValueError(f"Mapping '{mapping_name}' not found")
        mapping = self.mappings[mapping_name]
        result = {}
        for field_config in mapping.get('fields', []):
            source_field = field_config.get('source')
            target_field = field_config.get('target')
            # 获取源数据
            value = self._get_nested_value(source_data, source_field)
            # 应用转换
            if 'transformer' in field_config:
                transformer = self.transformers.get(field_config['transformer'])
                if transformer:
                    value = transformer(value, **field_config.get('params', {}))
            # 验证
            if 'validator' in field_config:
                validator = self.validators.get(field_config['validator'])
                if validator and not validator(value):
                    if field_config.get('required'):
                        raise ValueError(f"Validation failed for {target_field}")
                    continue
            # 处理默认值
            if value is None and 'default' in field_config:
                value = field_config['default']
            # 设置目标字段
            self._set_nested_value(result, target_field, value)
        return result
    def _get_nested_value(self, data: Dict, path: str) -> Any:
        """获取嵌套字段值"""
        keys = path.split('.')
        value = data
        for key in keys:
            if isinstance(value, dict) and key in value:
                value = value[key]
            else:
                return None
        return value
    def _set_nested_value(self, data: Dict, path: str, value: Any):
        """设置嵌套字段值"""
        keys = path.split('.')
        current = data
        for key in keys[:-1]:
            if key not in current:
                current[key] = {}
            current = current[key]
        current[keys[-1]] = value
# 映射配置文件 (mapping_config.yaml)
"""
mappings:
  user_to_customer:
    fields:
      - source: user.id
        target: customer.id
        transformer: int
      - source: user.name
        target: customer.full_name
        transformer: trim
      - source: user.email
        target: customer.email
        validator: email
        required: true
      - source: user.phone
        target: customer.phone
        transformer: format_phone
        default: "000-0000-0000"
      - source: user.birth_date
        target: customer.age
        transformer: calculate_age
      - source: user.address
        target: customer.address
        transformer: combine_address
"""
# 使用示例
mapper = AutoFieldMapper('mapping_config.yaml')
# 注册转换函数
mapper.register_transformer('int', int)
mapper.register_transformer('trim', lambda x: x.strip() if x else None)
mapper.register_transformer('format_phone', lambda x: f"+86-{x}" if x else None)
mapper.register_transformer('calculate_age', lambda x: 
    datetime.now().year - datetime.strptime(x, "%Y-%m-%d").year if x else None)
# 注册验证器
mapper.register_validator('email', lambda x: 
    bool(re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', x)) if x else False)
# 执行映射
source_data = {
    "user": {
        "id": "1001",
        "name": "  张三  ",
        "email": "zhangsan@example.com",
        "phone": "13800138000",
        "birth_date": "1990-01-15",
        "address": {
            "street": "科技园路",
            "city": "深圳"
        }
    }
}
result = mapper.map_fields(source_data, "user_to_customer")
print(json.dumps(result, indent=2, ensure_ascii=False))

批量映射处理

# batch_mapper.py
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
import logging
class BatchFieldMapper:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.mapper = AutoFieldMapper()
        self.batch_size = 1000
    def map_dataframe(self, df: pd.DataFrame, mapping_name: str) -> pd.DataFrame:
        """映射DataFrame数据"""
        records = df.to_dict('records')
        mapped_records = self.map_batch(records, mapping_name)
        return pd.DataFrame(mapped_records)
    def map_batch(self, records: list, mapping_name: str) -> list:
        """批量映射处理"""
        results = []
        errors = []
        # 分批处理
        for i in range(0, len(records), self.batch_size):
            batch = records[i:i + self.batch_size]
            try:
                # 单线程或线程池处理
                with ThreadPoolExecutor(max_workers=4) as executor:
                    futures = [
                        executor.submit(self.mapper.map_fields, record, mapping_name)
                        for record in batch
                    ]
                    for future in futures:
                        try:
                            result = future.result()
                            results.append(result)
                        except Exception as e:
                            errors.append(str(e))
                            self.logger.error(f"Mapping error: {e}")
            self.logger.info(f"Processed batch {i//self.batch_size + 1}")
        if errors:
            self.logger.warning(f"Total errors: {len(errors)}")
        return results
# 使用示例
mapper = BatchFieldMapper()
# 批量处理CSV文件
df = pd.read_csv('source_data.csv')
mapped_df = mapper.map_dataframe(df, 'user_to_customer')
mapped_df.to_csv('target_data.csv', index=False)

最佳实践建议

配置管理

# 使用配置文件管理映射
mappings:
  - name: customer_sync
    description: "客户数据同步映射"
    version: "1.0.0"
    source:
      table: "source_users"
      connection: "source_db"
    target:
      table: "target_customers"
      connection: "target_db"
    fields:
      - source: "id"
        target: "customer_id"
        type: "integer"
        required: true
      - source: "email"
        target: "email_address"
        type: "string"
        transform: "lowercase"
        validation: "email_format"
        on_null: "skip_record"

错误处理

class MappingError(Exception):
    """映射错误基类"""
    pass
class ValidationError(MappingError):
    """验证错误"""
    pass
class TransformationError(MappingError):
    """转换错误"""
    pass
def safe_mapping(source_data, mapping_func):
    """安全的映射处理"""
    try:
        result = mapping_func(source_data)
        return result, None
    except ValidationError as e:
        return None, f"Validation error: {e}"
    except TransformationError as e:
        return None, f"Transformation error: {e}"
    except Exception as e:
        return None, f"Unexpected error: {e}"

编写映射脚本时,建议：

1. 使用配置文件管理映射规则
2. 实现完善的错误处理机制
3. 添加日志记录便于追踪
4. 支持批量处理和并行处理
5. 保持映射逻辑的可测试性

这样就可以灵活处理各种表字段映射需求了。