将DataFrame列映射到目标表，缺失列设为空的Pythonic实现方案

从私有API爬取数据并转换为DataFrame（命名为raw_df），需要将其转换为指定结构的DataFrame：

• 仅保留12个指定列
• 重命名表头并设置对应数据类型
• 当raw_df缺失所需列时，目标DataFrame对应列设为空

当前使用try-except的实现不够优雅，寻求更Pythonic的写法。原代码如下：

df = pd.DataFrame({"ID de respuesta": pd.Series(dtype='int'),
                   "TiendaID": pd.Series(dtype='int'),
                   "Dirección IP": pd.Series(dtype='str'),
                   "Marca de tiempo (dd/mm/yyyy)": pd.Series(dtype='float'),
                   "Tiempo necesario para completar (segundos)": pd.Series(dtype='int'),
                   "Latitud": pd.Series(dtype='float'),
                   "Longitud": pd.Series(dtype='float'),
                   "¿El vendedor te recomendó algún producto adicional?": pd.Series(dtype='int'),
                   "¿Cuán probable es que recomiendes las tiendas Sx a un familiar o amigo?": pd.Series(dtype='int'),
                   "¿Por qué nos felicitas?": pd.Series(dtype='str'),
                   "¿En qué debemos mejorar?": pd.Series(dtype='str'),
                   "¿En qué fallamos?": pd.Series(dtype='str')})

df["ID de respuesta"] = raw_df["responseID"]
df["TiendaID"] = raw_df["surveyID"]
df["Dirección IP"] = raw_df["ipAddress"]
df["Marca de tiempo (dd/mm/yyyy)"] = raw_df["timestamp"]
df["Tiempo necesario para completar (segundos)"] = raw_df["timeTaken"]
df["Latitud"] = raw_df["location.latitude"]
df["Longitud"] = raw_df["location.longitude"]
df["¿El vendedor te recomendó algún producto adicional?"] = raw_df["¿El vendedor te recomendó algún producto adicional? "]
df["¿Cuán probable es que recomiendes las tiendas Sx a un familiar o amigo?"] = raw_df["¿Cuán probable es que recomiendes las tiendas Sx a un familiar o amigo?"]

try:
    df["¿Por qué nos felicitas?"] = raw_df["¿Por qué nos felicitas?"]
    df["¿En qué debemos mejorar?"] = raw_df["¿En qué debemos mejorar?"]
    df["¿En qué fallamos?"] = raw_df["¿En qué fallamos?"]
except Exception:
    pass 

核心思路是用列映射字典统一管理目标列名、原列名和数据类型，替代分散的赋值和try-except，让代码更易维护、逻辑更清晰。

方法一：循环赋值（直观易读） #

先定义一个字典绑定所有列的配置，再循环处理每一列，自动判断原列是否存在：

import pandas as pd

# 列映射字典：键=目标列名，值=(原DataFrame列名, 数据类型)
column_configs = {
    "ID de respuesta": ("responseID", "int"),
    "TiendaID": ("surveyID", "int"),
    "Dirección IP": ("ipAddress", "str"),
    "Marca de tiempo (dd/mm/yyyy)": ("timestamp", "float"),
    "Tiempo necesario para completar (segundos)": ("timeTaken", "int"),
    "Latitud": ("location.latitude", "float"),
    "Longitud": ("location.longitude", "float"),
    "¿El vendedor te recomendó algún producto adicional?": ("¿El vendedor te recomendó algún producto adicional? ", "int"),
    "¿Cuán probable es que recomiendes las tiendas Sx a un familiar o amigo?": ("¿Cuán probable es que recomiendes las tiendas Sx a un familiar o amigo?", "int"),
    "¿Por qué nos felicitas?": ("¿Por qué nos felicitas?", "str"),
    "¿En qué debemos mejorar?": ("¿En qué debemos mejorar?", "str"),
    "¿En qué fallamos?": ("¿En qué fallamos?", "str")
}

# 初始化空的目标DataFrame
df = pd.DataFrame()

# 遍历配置，处理每一列
for target_col, (source_col, dtype) in column_configs.items():
    # 原列存在则赋值并转类型，否则生成空系列
    df[target_col] = raw_df[source_col].astype(dtype) if source_col in raw_df.columns else pd.Series(dtype=dtype)

方法二：用reindex和rename简化（更高效） #

如果需要保持列顺序一致，也可以先提取存在的列并重命名，再补充缺失列：

# 拆分映射关系：原列名→目标列名，目标列名→数据类型
source_to_target = {source: target for target, (source, _) in column_configs.items()}
target_to_dtype = {target: dtype for target, (_, dtype) in column_configs.items()}

# 提取raw_df中存在的列，并重命名
existing_source_cols = [col for col in source_to_target.keys() if col in raw_df.columns]
df = raw_df[existing_source_cols].rename(columns=source_to_target)

# 补充缺失的列并设置对应类型
df = df.reindex(columns=column_configs.keys())
for col in df.columns:
    if df[col].isna().all():  # 仅对空列设置类型
        df[col] = df[col].astype(target_to_dtype[col])

方案优势 #

• 可维护性强：所有列的配置集中在字典中，修改、新增列只需调整字典，无需改动业务代码
• 逻辑清晰：无需依赖try-except捕获异常，通过显式判断列是否存在处理缺失情况
• 类型统一：数据类型和列配置绑定，避免分散设置导致的类型错误

内容的提问来源于stack exchange，提问作者Irving Pérez