本文目录导读：
我来详细解释如何编写不同表字段映射脚本,包含多种常见场景和实现方式。
基础字段映射脚本
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}"
编写映射脚本时,建议：
- 使用配置文件管理映射规则
- 实现完善的错误处理机制
- 添加日志记录便于追踪
- 支持批量处理和并行处理
- 保持映射逻辑的可测试性
这样就可以灵活处理各种表字段映射需求了。
