Python中如何编写Adjust log清洗脚本？学习《Python Cookbook》后的实践分享

不知道 V 站有多少同学的公司是使用 Adjust 作为出海应用的第三方数据检测平台的。在平时的使用过程中经常需要清洗一下从 Adjust 的 callback log(可是，我只是一只可怜的市场投放)。刚好在学习《 Python cookbook 》，就把其中关于生成器的部分拿了出来写了这个脚本(我连部分注释都没有修改)，不知道其他公司都是怎么处理 adjust 的数据的？期待能和其他使用 adjust 的公司的技术大大交流一下～～～。最后当然是放上渣代码 渣代码


Python中如何编写Adjust log清洗脚本？学习《Python Cookbook》后的实践分享

1 回复

帖子回复：

看了你的标题，我理解你是想分享一个实践，但问题本身是“如何编写Adjust log清洗脚本”。结合《Python Cookbook》的风格，这类数据处理任务的核心通常是：读取、解析、转换、输出。Adjust的日志通常是CSV或JSON格式，这里我假设是CSV，并给你一个可直接运行、扩展的骨架脚本。

import pandas as pd
import re
from datetime import datetime

def clean_adjust_log(input_file, output_file):
    """
    清洗Adjust日志的核心函数。
    思路：
    1. 读取原始数据（这里用pandas，处理大量数据时考虑chunksize或Dask）。
    2. 解析和清洗关键字段（如时间戳、设备ID、事件名）。
    3. 过滤无效数据（如空值、测试数据）。
    4. 转换数据格式（如统一时间格式、枚举值映射）。
    5. 保存清洗后的结果。
    """
    # 1. 读取数据
    # 假设日志是CSV格式，用逗号分隔。根据实际情况调整分隔符或使用`read_json`
    df = pd.read_csv(input_file)
    
    # 打印初始信息，方便调试
    print(f"原始数据行数: {len(df)}")
    print(f"原始字段: {df.columns.tolist()}")
    
    # 2. 清洗关键字段
    # 示例：统一时间戳格式（假设原始字段为`created_at`，格式可能不统一）
    if 'created_at' in df.columns:
        # 尝试解析日期，errors='coerce'将解析失败的设为NaT
        df['created_at'] = pd.to_datetime(df['created_at'], errors='coerce')
        # 删除无法解析时间的行（根据需求调整）
        df = df.dropna(subset=['created_at'])
    
    # 示例：清洗设备ID（假设字段为`device_id`，去除前后空格）
    if 'device_id' in df.columns:
        df['device_id'] = df['device_id'].astype(str).str.strip()
        # 过滤掉设备ID为'null'、'unknown'或空字符串的行
        invalid_ids = ['null', 'unknown', '']
        df = df[~df['device_id'].isin(invalid_ids)]
    
    # 示例：清洗事件名（假设字段为`event_name`，统一为小写）
    if 'event_name' in df.columns:
        df['event_name'] = df['event_name'].astype(str).str.lower().str.strip()
    
    # 3. 过滤无效数据
    # 删除所有列都为NaN的行
    df = df.dropna(how='all')
    # 删除重复行（根据关键字段去重，这里用`device_id`和`created_at`示例）
    key_columns = ['device_id', 'created_at']
    if all(col in df.columns for col in key_columns):
        df = df.drop_duplicates(subset=key_columns, keep='first')
    
    # 4. 转换数据格式（根据业务需求添加）
    # 例如：将某个数值字段除以100（如货币单位转换）
    # if 'revenue' in df.columns:
    #     df['revenue'] = df['revenue'] / 100.0
    
    # 5. 保存清洗后的数据
    df.to_csv(output_file, index=False)
    print(f"清洗完成！有效数据行数: {len(df)}")
    print(f"已保存到: {output_file}")

# 使用示例
if __name__ == "__main__":
    input_file = "raw_adjust_log.csv"  # 替换为你的输入文件路径
    output_file = "cleaned_adjust_log.csv"  # 替换为输出文件路径
    clean_adjust_log(input_file, output_file)

关键点解释：

  • pandas是核心：它提供了read_csv、to_datetime、dropna等高效方法，比手动写循环解析快得多。
  • 错误处理：比如pd.to_datetime(..., errors='coerce')，避免因个别格式错误导致整个脚本中断。
  • 可扩展性：你可以在清洗函数中添加更多步骤，比如用正则表达式提取用户ID、映射事件名到标准值等。
  • 《Python Cookbook》风格：这本书强调实用、可复用的代码块，这个函数就是一个模板，你可以根据实际日志结构调整字段名和清洗逻辑。

一句话建议： 先小样本测试，确保每步清洗都符合预期，再处理全量数据。

回到顶部