游戏流失数据预处理求助：ID匹配异常与多压缩文件处理

可能的异常原因 #

• tar包解压不完整，导致部分csv.gz文件损坏或缺失
• 不同csv.gz文件中ID字段类型不一致（如字符串/数值混合），合并后出现精度丢失或类型冲突
• 读取时使用error_bad_lines=False直接跳过错误行，遗漏含有效ID的数据
• 各文件列名存在大小写、空格差异，导致ID列被误过滤
• 多文件中重复ID未做统一处理，引发匹配冲突

分步处理方案 #

1. 确保tar包完整解压 #

避免手动解压可能出现的损坏问题，用Pythontarfile模块标准化解压流程：

import tarfile
import os

tar_path = "your_data.tar"  # 替换为你的tar包路径
extract_dir = "data/csv"
os.makedirs(extract_dir, exist_ok=True)

with tarfile.open(tar_path, 'r') as tar:
    if not tar.is_tarfile(tar_path):
        raise ValueError("输入文件不是有效的tar包")
    # 校验并解压所有文件
    tar.extractall(path=extract_dir)

2. 统一ID字段类型 #

强制将ID列转为字符串类型，避免大整数ID因数值精度丢失出现异常：

# 假设ID列名为user_id，根据实际字段名修改
df = pd.read_csv(filename, compression='gzip',
                 dtype={'user_id': str},  # 强制指定字符串类型
                 usecols=selected_cols)

3. 替换错误行处理逻辑 #

新版本pandas已弃用error_bad_lines，改用on_bad_lines捕获错误行而非直接跳过，方便排查：

# 自定义错误行处理函数，记录异常行到日志
def log_bad_lines(line):
    with open('bad_lines.log', 'a', encoding='utf-8') as f:
        f.write(','.join(line) + '\n')
    return None  # 返回None跳过该行，也可返回修正后的行

df = pd.read_csv(filename, compression='gzip',
                 dtype={'user_id': str},
                 on_bad_lines=log_bad_lines,  # 替换为警告或自定义处理
                 usecols=selected_cols)

4. 标准化列名 #

统一列名格式，避免因大小写、空格差异漏选ID列：

# 读取原始列名并标准化（小写+去首尾空格）
raw_headers = pd.read_csv(filename, nrows=1, compression='gzip').columns.tolist()
standardized_headers = [col.strip().lower() for col in raw_headers]
# 同步标准化需要移除的列名
standardized_removable = [col.strip().lower() for col in removable_columns]
# 筛选需要保留的列（匹配标准化后的名称）
selected_cols = [raw_col for raw_col, std_col in zip(raw_headers, standardized_headers) 
                 if std_col not in standardized_removable]

# 读取数据后统一列名格式
df = pd.read_csv(filename, compression='gzip',
                 dtype={'user_id': str},
                 on_bad_lines=log_bad_lines,
                 usecols=selected_cols)
df.columns = standardized_headers

5. 数据校验与异常记录 #

每个文件读取后，检查ID列的完整性，记录异常文件：

appended_data = []
error_records = []

for filename in my_file_list:
    # 标准化列名并筛选列（代码同步骤4）
    raw_headers = pd.read_csv(filename, nrows=1, compression='gzip').columns.tolist()
    standardized_headers = [col.strip().lower() for col in raw_headers]
    standardized_removable = [col.strip().lower() for col in removable_columns]
    selected_cols = [raw_col for raw_col, std_col in zip(raw_headers, standardized_headers) 
                     if std_col not in standardized_removable]
    
    # 读取数据
    df = pd.read_csv(filename, compression='gzip',
                     dtype={'user_id': str},
                     on_bad_lines=log_bad_lines,
                     usecols=selected_cols)
    df.columns = standardized_headers
    
    # 检查ID列是否存在
    if 'user_id' not in df.columns:
        error_records.append(f"{filename}: 缺失user_id字段")
        continue
    
    # 检查ID缺失值
    missing_count = df['user_id'].isnull().sum()
    if missing_count > 0:
        error_records.append(f"{filename}: {missing_count}条数据缺失user_id")
        df = df.dropna(subset=['user_id'])  # 删除缺失行，或根据业务标记
    
    appended_data.append(df)

# 合并所有数据
final_df = pd.concat(appended_data, ignore_index=True)

# 写入错误日志
if error_records:
    with open('data_errors.log', 'w', encoding='utf-8') as f:
        f.write('\n'.join(error_records))

6. 重复ID处理 #

根据业务需求处理多文件中的重复ID：

# 示例1：保留最新登录记录（假设存在login_time字段）
final_df = final_df.sort_values('login_time', ascending=False).drop_duplicates('user_id', keep='first')

# 示例2：聚合重复ID的行为数据（如统计游戏时长总和）
# final_df = final_df.groupby('user_id').agg(
#     total_game_time=('game_time', 'sum'),
#     last_login=('login_time', 'max')
# ).reset_index()

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