如何将Pandas分类列因子化并生成映射表？大表适配DASK方案咨询

我正在处理SQL Server上一张大型非规范化表（10列×1.3亿行），以下为数据示例：

import pandas as pd
import numpy as np
data = pd.DataFrame({
    'status' :  ['pending', 'pending','pending', 'canceled','canceled','canceled', 'confirmed', 'confirmed','confirmed'],
    'clientId' : ['A', 'B', 'C', 'A', 'D', 'C', 'A', 'B','C'],
    'partner' :  ['A', np.nan,'C', 'A',np.nan,'C', 'A', np.nan,'C'],
    'product' : ['afiliates', 'pre-paid', 'giftcard','afiliates', 'pre-paid', 'giftcard','afiliates', 'pre-paid', 'giftcard'],
    'brand' : ['brand_1', 'brand_2', 'brand_3','brand_1', 'brand_2', 'brand_3','brand_1', 'brand_3', 'brand_3'],
    'gmv' : [100,100,100,100,100,100,100,100,100]})

data = data.astype({'partner':'category','status':'category','product':'category', 'brand':'category'})

如您所见，其中多列为分类/字符串类型，可进行factorize（替换为小整数标识，用于关联查询）。请问是否有简便方法从每个分类列提取映射表，并将主表因子化，以提升单查询的字节传输速度？是否有合适的工具库可用？

预期输出如下：

data = pd.DataFrame({
    'status' :  ['1', '1','1', '2','2','2', '3', '3','3'],
    'clientId' : ['1', '2', '3', '1', '4', '3', '1', '2','3'],
    'partner' :  ['A', np.nan,'C', 'A',np.nan,'C', 'A', np.nan,'C'],
    'product' : ['afiliates', 'pre-paid', 'giftcard','afiliates', 'pre-paid', 'giftcard','afiliates', 'pre-paid', 'giftcard'],
    'brand' : ['brand_1', 'brand_2', 'brand_3','brand_1', 'brand_2', 'brand_3','brand_1', 'brand_3', 'brand_3'],
    'gmv' : [100,100,100,100,100,100,100,100,100]})

status_df = {1 : 'pending', 2:'canceled', 3:'confirmed'} 
clientid = {1 : 'A', 2:'B', 3:'C', 4:'D'}

以此类推！

额外问题：由于表数据量极大，我可能需要用DASK实现，求相关解决方案。

解决方案：Pandas 实现 #

核心逻辑 #

利用Pandas的factorize()方法或分类列的cat属性，快速完成列编码并提取映射关系，既保证效率又能直接生成所需的整数编码和映射表。

代码实现 #

import pandas as pd
import numpy as np

# 原始示例数据
data = pd.DataFrame({
    'status' :  ['pending', 'pending','pending', 'canceled','canceled','canceled', 'confirmed', 'confirmed','confirmed'],
    'clientId' : ['A', 'B', 'C', 'A', 'D', 'C', 'A', 'B','C'],
    'partner' :  ['A', np.nan,'C', 'A',np.nan,'C', 'A', np.nan,'C'],
    'product' : ['afiliates', 'pre-paid', 'giftcard','afiliates', 'pre-paid', 'giftcard','afiliates', 'pre-paid', 'giftcard'],
    'brand' : ['brand_1', 'brand_2', 'brand_3','brand_1', 'brand_2', 'brand_3','brand_1', 'brand_3', 'brand_3'],
    'gmv' : [100,100,100,100,100,100,100,100,100]})

# 转为category类型（已转换的可跳过此步骤）
data = data.astype({'partner':'category','status':'category','product':'category', 'brand':'category'})

# 指定需要编码的列
encode_cols = ['status', 'clientId', 'partner', 'product', 'brand']

# 存储所有列的映射表
mapping_dict = {}

# 遍历列执行编码并生成映射
for col in encode_cols:
    if data[col].dtype == 'category':
        # 分类列直接用cat属性获取编码和标签，编码从1开始避免与NaN冲突
        codes = data[col].cat.codes + 1
        categories = data[col].cat.categories
        mapping = dict(zip(codes.unique(), categories))
    else:
        # 非分类列用factorize生成编码和标签
        codes, categories = pd.factorize(data[col], sort=True)
        codes = codes + 1
        mapping = dict(zip(range(1, len(categories)+1), categories))
    
    # 更新主表列为编码值（转为字符串类型匹配示例输出，也可保留整数）
    data[col] = codes.astype(str)
    # 保存当前列的映射表
    mapping_dict[f"{col}_mapping"] = mapping

# 输出结果
print("编码后主表:\n", data)
print("\nstatus映射表:", mapping_dict['status_mapping'])
print("clientId映射表:", mapping_dict['clientId_mapping'])

关键说明 #

• 编码从1开始，避免与Pandas默认的NaN编码（-1）混淆，后续还原或关联查询更易处理
• 已转为category类型的列处理效率远高于普通字符串列，建议提前转换
• 所有映射表统一存储在字典中，便于后续批量还原或与其他表关联

解决方案：Dask 实现 #

针对1.3亿行的超大规模数据，Dask通过分块并行处理避免内存溢出，核心逻辑与Pandas一致，但需先生成全局统一映射表，确保分块编码一致性。

代码实现 #

import dask.dataframe as dd
import pandas as pd

# 从SQL Server读取数据，按100万行分块（可根据内存调整chunksize）
conn_str = "mssql+pyodbc://username:password@server/database?driver=ODBC+Driver+17+for+SQL+Server"
query = "SELECT * FROM large_table"
dask_df = dd.read_sql(query, conn_str, chunksize=1_000_000)

# 指定需要编码的列
encode_cols = ['status', 'clientId', 'partner', 'product', 'brand']

# 生成全局统一映射表，避免分块编码不一致
def get_global_mappings(df, cols):
    mappings = {}
    for col in cols:
        # 计算全量唯一值
        unique_vals = df[col].unique().compute()
        # 生成编码从1开始的映射
        mappings[col] = dict(zip(range(1, len(unique_vals)+1), unique_vals))
    return mappings

global_mappings = get_global_mappings(dask_df, encode_cols)

# 定义分块编码函数
def encode_chunk(chunk, mappings, cols):
    for col in cols:
        # 用全局映射替换原始值为编码，忽略NaN
        chunk[col] = chunk[col].map(mappings[col], na_action='ignore').astype(str)
    return chunk

# 对所有分块应用编码
encoded_dask_df = dask_df.map_partitions(encode_chunk, global_mappings, encode_cols)

# 可选：将编码后的数据写入SQL Server或存储为Parquet（推荐Parquet，压缩比高传输快）
encoded_dask_df.to_sql("encoded_large_table", conn_str, if_exists="replace", index=False)
# encoded_dask_df.to_parquet("encoded_large_table.parquet", compression='snappy')

# 输出全局映射表
print("全局映射表:\n", global_mappings)

关键说明 #

• 先生成全局映射表，确保不同分块的同一标签对应相同编码
• 使用map_partitions并行处理分块，效率远高于单线程处理
• 推荐将编码后的数据存储为Parquet格式，大幅降低存储体积和传输字节数，后续查询速度更快

内容的提问来源于stack exchange，提问作者FábioRB