如何编写转换数据表字段映射脚本

本文目录导读：

1. 基础字段映射脚本结构
2. 复杂映射场景
3. YAML配置映射
4. SQL字段映射
5. 最佳实践建议
6. 测试示例

我来详细介绍编写数据表字段映射脚本的方法,包含多个实际场景的示例。

基础字段映射脚本结构

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()

编写字段映射脚本的关键是：

1. 明确映射规则：清晰定义源字段和目标字段的对应关系
2. 处理异常情况：考虑空值、类型不匹配等问题
3. 保持可配置性：使用配置文件而非硬编码
4. 添加日志记录：便于调试和审计
5. 进行充分测试：验证映射结果的准确性

根据实际需求选择合适的实现方式,可以是简单的Python脚本，也可以结合YAML配置或SQL实现更复杂的映射逻辑。