数据管道项目中Python动态类型转换（Unicode转Python对象）求助

Hey there! Let's tackle this dynamic type conversion issue you're facing in your data pipeline project. I’ve dealt with similar scenarios before, so here’s a practical approach to solve it:

The core idea here is to map the type names from your metadata to actual Python type objects (or custom conversion functions), then apply that mapping to your user data rows. Here's a step-by-step implementation:

1. Create a Type Mapping Dictionary #

First, define a dictionary that translates the type strings from your metadata (like "float", "int") to Python's built-in types (or custom parsers for complex types like dates):

from datetime import datetime

# Custom parser for date strings (adjust the format to match your data)
def parse_date(date_str):
    return datetime.strptime(date_str, "%Y-%m-%d")

# Map metadata type names to Python types/parsers
type_mapping = {
    "int": int,
    "float": float,
    "str": str,
    "bool": bool,
    "date": parse_date
}

2. Process Data with Metadata-Driven Conversion #

Assume you've loaded your column metadata and user data (e.g., as dictionaries or rows from a database/file). Here's how to apply the conversion:

# Example metadata (replace with data loaded from your metadata source)
column_metadata = [
    {"column_name": "user_id", "data_type": "int"},
    {"column_name": "account_balance", "data_type": "float"},
    {"column_name": "is_subscribed", "data_type": "bool"},
    {"column_name": "signup_date", "data_type": "date"},
    {"column_name": "username", "data_type": "str"}
]

# Example raw user data row (replace with your actual user data)
raw_user_data = {
    "user_id": "9876",
    "account_balance": "1250.75",
    "is_subscribed": "False",
    "signup_date": "2023-01-15",
    "username": "jane_smith"
}

processed_data = {}
for col_meta in column_metadata:
    col_name = col_meta["column_name"]
    raw_value = raw_user_data[col_name]
    target_type = type_mapping.get(col_meta["data_type"])

    if not target_type:
        # Handle unknown types (fallback to string or log a warning)
        processed_data[col_name] = str(raw_value)
        print(f"Warning: Unknown data type '{col_meta['data_type']}' for column {col_name} — defaulting to string")
        continue

    try:
        # Special handling for boolean values (since bool("False") returns True in Python)
        if target_type is bool:
            processed_value = raw_value.strip().lower() == "true"
        else:
            processed_value = target_type(raw_value)
        
        processed_data[col_name] = processed_value
    except ValueError as e:
        # Handle conversion failures (e.g., invalid number strings, malformed dates)
        print(f"Error converting '{raw_value}' to {target_type.__name__} for column {col_name}: {str(e)}")
        # You can set a default value, skip the row, or log to a error file here
        processed_data[col_name] = None

print(processed_data)
# Output:
# {
#     'user_id': 9876,
#     'account_balance': 1250.75,
#     'is_subscribed': False,
#     'signup_date': datetime.datetime(2023, 1, 15, 0, 0),
#     'username': 'jane_smith'
# }

Key Notes for Production Use #

• Error Handling: Expand the try/except blocks to log errors to a file or monitoring system instead of just printing — this helps track dirty data issues in your pipeline.
• Metadata Consistency: Ensure your metadata uses consistent type names (e.g., don't mix "integer" and "int"). Add a step to normalize type names when loading metadata if needed.
• Complex Types: For nested structures or custom data types, extend the type_mapping with custom conversion functions that handle those cases.

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