本文目录导读：
我来详细介绍编写数据表字段映射脚本的方法,包含多个实际场景的示例。
基础字段映射脚本结构
Python示例 - 简单的字段映射
import pandas as pd
import json
class FieldMapper:
def __init__(self, mapping_config):
"""
初始化字段映射器
:param mapping_config: 字段映射配置字典
"""
self.mapping = mapping_config
def map_fields(self, df):
"""
执行字段映射转换
:param df: 源数据DataFrame
:return: 映射后的DataFrame
"""
mapped_data = {}
for target_field, source_config in self.mapping.items():
source_field = source_config.get('source_field')
transform_func = source_config.get('transform', lambda x: x)
if source_field in df.columns:
mapped_data[target_field] = df[source_field].apply(transform_func)
elif source_config.get('default') is not None:
mapped_data[target_field] = source_config['default']
return pd.DataFrame(mapped_data)
# 使用示例
mapping_config = {
'user_id': {
'source_field': 'id',
'transform': lambda x: f"USR_{x}"
},
'user_name': {
'source_field': 'name',
'transform': str.strip
},
'email': {
'source_field': 'email',
'transform': lambda x: x.lower() if pd.notna(x) else ''
},
'created_at': {
'source_field': 'register_date',
'transform': lambda x: pd.to_datetime(x).strftime('%Y-%m-%d')
},
'status': {
'source_field': None, # 没有对应字段
'default': 'active' # 使用默认值
}
}
# 创建映射器并转换数据
mapper = FieldMapper(mapping_config)
source_data = pd.DataFrame({
'id': [1, 2, 3],
'name': [' Alice ', 'Bob', 'Charlie'],
'email': ['ALICE@EXAMPLE.COM', 'BOB@EXAMPLE.COM', None],
'register_date': ['2023-01-15', '2023-02-20', '2023-03-25']
})
result = mapper.map_fields(source_data)
print(result)
复杂映射场景
多表关联映射
import pandas as pd
class AdvancedFieldMapper:
def __init__(self):
self.mappers = {}
def add_mapper(self, name, mapping_config):
"""添加映射配置"""
self.mappers[name] = mapping_config
def map_with_lookup(self, df, lookup_table, lookup_key, target_field):
"""
使用查找表进行字段映射
"""
mapping_dict = dict(zip(lookup_table[lookup_key], lookup_table[target_field]))
return df.map(mapping_dict)
def combine_fields(self, df, fields, separator='_'):
"""合并多个字段"""
return df[fields].apply(lambda x: separator.join(x.astype(str)), axis=1)
def conditional_mapping(self, df, condition_config):
"""
条件映射
condition_config = {
'target_field': 'approval_status',
'conditions': [
{'source': 'amount', 'condition': lambda x: x > 1000, 'value': '需要审批'},
{'source': 'amount', 'condition': lambda x: x <= 1000, 'value': '自动通过'},
],
'default': '待处理'
}
"""
target_field = condition_config['target_field']
result = pd.Series([condition_config['default']] * len(df))
for condition in condition_config['conditions']:
mask = df[condition['source']].apply(condition['condition'])
result[mask] = condition['value']
return result
类型转换映射
from datetime import datetime
import re
class TypeConverter:
"""类型转换器"""
@staticmethod
def string_to_date(date_str, format='%Y-%m-%d'):
"""字符串转日期"""
if pd.isna(date_str) or date_str == '':
return None
try:
return datetime.strptime(str(date_str), format)
except ValueError:
return None
@staticmethod
def string_to_number(value):
"""字符串转数字"""
if pd.isna(value) or value == '':
return 0.0
# 清理字符串（移除货币符号、逗号等）
cleaned = re.sub(r'[$,€£¥\s]', '', str(value))
try:
return float(cleaned)
except ValueError:
return 0.0
@staticmethod
def boolean_mapping(value, mapping={'yes': True, 'no': False, '1': True, '0': False}):
"""布尔值映射"""
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
return mapping.get(str(value).lower().strip(), False)
YAML配置映射
mapping_config.yaml
# 字段映射配置文件
mappings:
source_database: mysql
target_database: postgresql
batch_size: 1000
field_mappings:
# 简单字段映射
- source: user_id
target: customer_id
transform:
type: prefix
value: "CUS_"
- source: full_name
target: display_name
transform:
type: split_join
separator: " "
join_char: " "
# 字段合并
- target: address
combine:
- street_address
- city
- state
- zip_code
separator: ", "
# 查找映射
- source: country_code
target: country_name
lookup:
table: country_codes
key_column: code
value_column: name
# 条件映射
- target: risk_level
conditions:
- source: age
operator: ">"
value: 60
result: "high"
- source: age
operator: "between"
min: 18
max: 60
result: "medium"
default: "low"
Python读取YAML配置
import yaml
import pandas as pd
class YAMLFieldMapper:
def __init__(self, config_path):
with open(config_path, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
def apply_mapping(self, source_df):
"""应用YAML配置的映射"""
result = {}
for mapping in self.config['mappings']['field_mappings']:
target = mapping.get('target')
# 处理简单映射
if 'source' in mapping:
value = source_df[mapping['source']]
# 应用转换
if 'transform' in mapping:
value = self._apply_transform(value, mapping['transform'])
result[target] = value
# 处理字段合并
elif 'combine' in mapping:
fields = [source_df[field] for field in mapping['combine']]
result[target] = pd.concat(fields, axis=1).apply(
lambda x: mapping.get('separator', ', ').join(x.astype(str)),
axis=1
)
# 处理条件映射
elif 'conditions' in mapping:
result[target] = self._apply_conditions(source_df, mapping)
return pd.DataFrame(result)
def _apply_transform(self, series, transform_config):
"""应用转换规则"""
transform_type = transform_config.get('type')
if transform_type == 'prefix':
return series.apply(lambda x: f"{transform_config['value']}{x}")
elif transform_type == 'suffix':
return series.apply(lambda x: f"{x}{transform_config['value']}")
elif transform_type == 'split_join':
return series.apply(lambda x: transform_config['join_char'].join(
str(x).split(transform_config.get('separator', ' '))
))
return series
def _apply_conditions(self, df, condition_config):
"""应用条件映射"""
default = condition_config.get('default', None)
result = pd.Series([default] * len(df))
for condition in condition_config['conditions']:
source = condition['source']
operator = condition['operator']
if operator == '>':
mask = df[source] > condition['value']
elif operator == '<':
mask = df[source] < condition['value']
elif operator == '==':
mask = df[source] == condition['value']
elif operator == 'between':
mask = (df[source] >= condition['min']) & (df[source] <= condition['max'])
elif operator == 'in':
mask = df[source].isin(condition['values'])
result[mask] = condition['result']
return result
SQL字段映射
使用SQL进行字段映射
-- 创建映射函数
CREATE OR REPLACE FUNCTION map_user_fields()
RETURNS TABLE (
mapped_user_id VARCHAR,
mapped_full_name VARCHAR,
mapped_email VARCHAR,
mapped_created_at TIMESTAMP
) AS $$
BEGIN
RETURN QUERY
SELECT
-- 字段映射和转换
CONCAT('USR_', u.id::VARCHAR) AS mapped_user_id,
INITCAP(u.first_name || ' ' || u.last_name) AS mapped_full_name,
LOWER(u.email) AS mapped_email,
COALESCE(u.created_at, CURRENT_TIMESTAMP) AS mapped_created_at
FROM
source_table u
WHERE
u.is_active = TRUE;
END;
$$ LANGUAGE plpgsql;
-- 使用CASE语句进行条件映射
SELECT
id,
CASE
WHEN age >= 60 THEN 'Senior'
WHEN age >= 18 THEN 'Adult'
WHEN age > 0 THEN 'Minor'
ELSE 'Unknown'
END AS age_group,
CASE
WHEN status IN ('active', 'verified') THEN 'Approved'
WHEN status = 'pending' THEN 'Pending'
ELSE 'Rejected'
END AS approval_status
FROM users;
最佳实践建议
错误处理和日志
import logging
from functools import wraps
def log_mapping_errors(func):
"""装饰器：记录映射错误"""
@wraps(func)
def wrapper(*args, **kwargs):
logger = logging.getLogger('field_mapper')
try:
return func(*args, **kwargs)
except Exception as e:
logger.error(f"Field mapping error in {func.__name__}: {str(e)}")
raise
return wrapper
class RobustMapper:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.error_records = []
@log_mapping_errors
def safe_mapping(self, df, mapping):
"""安全的字段映射，记录错误"""
results = []
for idx, row in df.iterrows():
try:
mapped_row = self._map_single_row(row, mapping)
results.append(mapped_row)
except Exception as e:
self.error_records.append({
'row_index': idx,
'error': str(e),
'data': row.to_dict()
})
self.logger.warning(f"Failed to map row {idx}: {e}")
continue
return pd.DataFrame(results)
def _map_single_row(self, row, mapping):
"""映射单行数据"""
mapped = {}
for target, config in mapping.items():
source = config.get('source')
if source and source in row.index:
value = row[source]
# 类型转换
if 'type' in config:
value = self._type_cast(value, config['type'])
# 数据验证
if 'validation' in config:
self._validate(value, config['validation'])
mapped[target] = value
elif 'default' in config:
mapped[target] = config['default']
else:
raise ValueError(f"Missing source field: {source}")
return mapped
def _type_cast(self, value, target_type):
"""类型转换"""
try:
if target_type == 'int':
return int(value)
elif target_type == 'float':
return float(value)
elif target_type == 'string':
return str(value)
elif target_type == 'date':
return pd.to_datetime(value)
except (ValueError, TypeError) as e:
self.logger.warning(f"Type cast error: {e}")
return value
def _validate(self, value, validation_rules):
"""数据验证"""
for rule, param in validation_rules.items():
if rule == 'required' and param and pd.isna(value):
raise ValueError(f"Required field is missing")
elif rule == 'max_length' and len(str(value)) > param:
raise ValueError(f"Value exceeds max length {param}")
elif rule == 'pattern' and not re.match(param, str(value)):
raise ValueError(f"Value does not match pattern {param}")
def get_error_report(self):
"""获取错误报告"""
return pd.DataFrame(self.error_records)
性能优化
import numpy as np
from concurrent.futures import ThreadPoolExecutor
class OptimizedMapper:
def __init__(self, batch_size=10000):
self.batch_size = batch_size
def parallel_mapping(self, df, mapping_func):
"""并行处理映射"""
batches = [df[i:i+self.batch_size]
for i in range(0, len(df), self.batch_size)]
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(mapping_func, batches))
return pd.concat(results, ignore_index=True)
def vectorized_mapping(self, df, mapping_rules):
"""向量化映射（避免循环）"""
result = pd.DataFrame()
for target, rules in mapping_rules.items():
if 'source' in rules:
# 使用向量化操作
source_col = df[rules['source']]
if 'transform' in rules:
result[target] = np.vectorize(rules['transform'])(source_col)
else:
result[target] = source_col
return result
测试示例
import unittest
import pandas as pd
class TestFieldMapper(unittest.TestCase):
def setUp(self):
self.mapper = FieldMapper({
'user_id': {'source_field': 'id', 'transform': lambda x: f"USR_{x}"},
'user_name': {'source_field': 'name', 'transform': str.strip}
})
self.test_data = pd.DataFrame({
'id': [1, 2, 3],
'name': [' Alice ', 'Bob', 'Charlie']
})
def test_basic_mapping(self):
result = self.mapper.map_fields(self.test_data)
self.assertIn('user_id', result.columns)
self.assertIn('user_name', result.columns)
self.assertEqual(result['user_id'][0], 'USR_1')
self.assertEqual(result['user_name'][0], 'Alice')
def test_missing_source(self):
mapper = FieldMapper({
'new_field': {'source_field': 'non_existent', 'default': 'DEFAULT'}
})
result = mapper.map_fields(self.test_data)
self.assertEqual(result['new_field'][0], 'DEFAULT')
if __name__ == '__main__':
unittest.main()
编写字段映射脚本的关键是：
- 明确映射规则：清晰定义源字段和目标字段的对应关系
- 处理异常情况：考虑空值、类型不匹配等问题
- 保持可配置性：使用配置文件而非硬编码
- 添加日志记录：便于调试和审计
- 进行充分测试：验证映射结果的准确性
根据实际需求选择合适的实现方式,可以是简单的Python脚本，也可以结合YAML配置或SQL实现更复杂的映射逻辑。
