当前位置：首页 > news >正文

Qwen2.5-Coder-1.5B代码生成实战：从零到一完成数据清洗脚本

news 2026/7/27 22:12:27

Qwen2.5-Coder-1.5B代码生成实战：从零到一完成数据清洗脚本

1. 为什么你需要一个懂代码的AI助手

如果你经常和数据打交道，一定遇到过这样的场景：业务部门发来一份格式混乱的Excel表格，里面有合并单元格、多余的空行、日期格式不统一，还有一堆需要手动计算的字段。你打开Python编辑器，准备写个清洗脚本，结果光是处理那些奇葩的格式问题就花了半天时间，最后写出来的代码又长又乱，下次遇到类似问题还得从头再来。

这就是为什么你需要Qwen2.5-Coder-1.5B这样的专业代码生成模型。它不是那种只会聊天、给出模糊建议的通用AI，而是经过5.5万亿代码token专门训练出来的“编程专家”。它只有1.5B参数，部署起来轻量快捷，但在生成Python、JavaScript、SQL等40多种语言的代码时，表现相当扎实。

今天我就带你用Qwen2.5-Coder-1.5B，从零开始完成一个真实的数据清洗任务。整个过程就像有个经验丰富的同事在旁边指导，你只需要描述清楚需求，它就能给出可运行的代码，而且质量比很多初级程序员写得还要好。

2. 快速上手：三分钟部署你的代码助手

2.1 环境准备：比你想的简单

很多人一听到“部署模型”就觉得麻烦，其实用CSDN星图镜像，整个过程比安装一个普通软件还简单。你不需要懂Docker，也不用配置复杂的Python环境。

首先访问CSDN星图镜像广场，找到Qwen2.5-Coder-1.5B的镜像。这个镜像已经预装好了所有依赖，包括模型文件、推理框架和Web界面，你只需要点击“一键部署”就行。

部署完成后，你会看到一个简洁的Web界面。左边是聊天窗口，右边是代码编辑区。整个界面设计得很直观，就像在用ChatGPT写代码一样自然。

2.2 第一次对话：让它理解你的需求

在输入框里，你可以用最自然的方式描述你的数据清洗需求。比如：

“我有一个CSV文件，里面是销售数据。需要做这些处理：第一列是日期，格式乱七八糟，有的写‘2024-01-15’，有的写‘1/15/2024’，需要统一成‘YYYY-MM-DD’格式。第二列是销售额，有些单元格是空的，需要用这一列的平均值填充。第三列是产品类别，里面有些拼写错误，比如‘Electronics’写成了‘Electornics’，需要自动纠正。最后，删除所有重复的行，然后按日期排序。”

点击发送，Qwen2.5-Coder-1.5B会在几秒钟内给你一个完整的Python脚本。它不只是给出代码片段，而是包含文件读取、数据处理、错误处理、结果保存的完整程序。

2.3 运行验证：看看效果怎么样

把生成的代码复制到你的Python环境里运行（或者直接在镜像提供的代码编辑区运行），它会告诉你处理进度，比如“读取了1000行数据”、“修正了15个拼写错误”、“删除了23条重复记录”。最后生成一个新的CSV文件，所有数据都变得干净整齐。

如果第一次运行有问题（比如你的文件路径不对），直接把错误信息复制给Qwen2.5-Coder-1.5B，它会立刻给出修复方案。这种交互式的调试体验，比在Stack Overflow上搜半天要高效得多。

3. 实战演练：一步步构建完整的数据清洗脚本

3.1 场景设定：一个真实的电商数据清洗任务

假设你在一家电商公司，市场部每天都会给你一份用户行为数据，需要清洗后才能用于分析。原始数据存在这些问题：

用户ID有的是数字，有的是字符串，格式不统一
行为时间戳有的是Unix时间戳，有的是“2024-03-15 14:30:00”这样的字符串
商品价格有的带美元符号，有的不带，有的还有逗号分隔（如“1,299.99”）
用户地域信息有的写省份，有的写城市，有的甚至写拼音
有些行数据不全，缺少关键字段

传统做法是写一堆正则表达式和条件判断，代码越写越长，维护起来头疼。现在我们用Qwen2.5-Coder-1.5B来优雅地解决。

3.2 第一步：让AI理解数据结构和清洗规则

不要一上来就让AI写完整代码。先让它理解你的数据长什么样，以及具体的清洗规则。你可以这样描述：

“我需要处理一个电商用户行为日志CSV文件，包含以下列：user_id, event_time, product_id, price, region, device_type。清洗规则如下：

user_id统一转为字符串，确保长度一致
event_time统一转为datetime对象，原始格式可能是Unix时间戳或‘YYYY-MM-DD HH:MM:SS’
price去掉美元符号和逗号，转为浮点数
region映射到标准省份名称，建立映射字典：{‘beijing’: ‘北京’, ‘shanghai’: ‘上海’…}
删除device_type为空的行
对每个user_id，只保留最近的一次事件请生成完整的Python清洗脚本，使用pandas库，包含详细的注释。”

Qwen2.5-Coder-1.5B会生成这样的代码框架：

import pandas as pd import numpy as np from datetime import datetime def clean_ecommerce_data(input_file, output_file): """ 清洗电商用户行为数据 参数: input_file: 输入CSV文件路径 output_file: 输出CSV文件路径 """ # 读取数据 df = pd.read_csv(input_file) print(f"原始数据行数: {len(df)}") # 1. 清洗user_id df['user_id'] = df['user_id'].astype(str).str.zfill(8) # 统一为8位字符串 # 2. 统一event_time格式 def parse_time(time_str): try: # 尝试解析为Unix时间戳 if str(time_str).isdigit(): return datetime.fromtimestamp(int(time_str)) # 尝试解析为字符串格式 return pd.to_datetime(time_str) except: return pd.NaT df['event_time'] = df['event_time'].apply(parse_time) # 3. 清洗price列 df['price'] = df['price'].replace({'\$': '', ',': ''}, regex=True) df['price'] = pd.to_numeric(df['price'], errors='coerce') # 4. 标准化region region_mapping = { 'beijing': '北京', 'bj': '北京', 'shanghai': '上海', 'sh': '上海', 'guangzhou': '广州', 'gz': '广州', 'shenzhen': '深圳', 'sz': '深圳' } df['region'] = df['region'].str.lower().map(region_mapping).fillna(df['region']) # 5. 删除device_type为空的行 df = df.dropna(subset=['device_type']) # 6. 每个用户只保留最近事件 df = df.sort_values('event_time').drop_duplicates('user_id', keep='last') # 保存结果 df.to_csv(output_file, index=False) print(f"清洗后数据行数: {len(df)}") print(f"数据已保存到: {output_file}") return df # 使用示例 if __name__ == "__main__": cleaned_df = clean_ecommerce_data('raw_data.csv', 'cleaned_data.csv')

3.3 第二步：处理复杂情况，让代码更健壮

上面的代码已经能处理大部分情况，但实际数据往往更“脏”。比如有些价格字段里混入了中文“元”，有些时间戳是毫秒级的，有些region字段是“北京市朝阳区”这样的详细地址。

这时候你可以继续和Qwen2.5-Coder-1.5B对话：

“上面的代码很好，但还需要增强：1. price字段可能包含‘元’字，比如‘1299元’，需要去掉并转为数字；2. 有些时间戳是13位毫秒级，需要处理；3. region字段需要从详细地址中提取省份；4. 增加数据质量报告，统计清洗前后变化。”

AI会给出增强版的代码，增加相应的处理逻辑：

# 增强的price清洗函数 def clean_price(price_val): if pd.isna(price_val): return np.nan price_str = str(price_val) # 移除货币符号和单位 price_str = price_str.replace('$', '').replace('￥', '').replace('元', '') # 移除千分位逗号 price_str = price_str.replace(',', '') try: return float(price_str) except: return np.nan # 增强的时间解析函数 def parse_timestamp(ts): ts_str = str(ts) if ts_str.isdigit(): # 判断是秒级还是毫秒级时间戳 if len(ts_str) == 13: # 毫秒级 return datetime.fromtimestamp(int(ts_str) / 1000) else: # 秒级 return datetime.fromtimestamp(int(ts_str)) else: return pd.to_datetime(ts, errors='coerce') # 从详细地址提取省份 def extract_province(address): if pd.isna(address): return np.nan address = str(address) provinces = ['北京', '上海', '广州', '深圳', '浙江', '江苏', '广东'] for province in provinces: if province in address: return province return address # 没匹配到就返回原值 # 生成数据质量报告 def generate_quality_report(df_before, df_after): report = { '原始数据行数': len(df_before), '清洗后行数': len(df_after), '删除行数': len(df_before) - len(df_after), 'user_id格式统一率': (df_after['user_id'].str.len() == 8).mean(), 'event_time解析成功率': df_after['event_time'].notna().mean(), 'price有效比例': df_after['price'].notna().mean(), 'region标准化比例': df_after['region'].isin(['北京', '上海', '广州', '深圳']).mean() } return report

3.4 第三步：添加测试和异常处理

好的数据清洗脚本不能只处理“理想情况”，还要能应对各种异常。让Qwen2.5-Coder-1.5B为你的脚本添加单元测试：

“为上面的清洗函数添加单元测试，覆盖以下场景：1. 输入文件不存在；2. 数据列缺失；3. 所有price都是无效值；4. 时间戳格式异常。使用pytest框架。”

生成的测试代码会是这样：

import pytest import pandas as pd from io import StringIO def test_file_not_found(): """测试文件不存在的情况""" with pytest.raises(FileNotFoundError): clean_ecommerce_data('nonexistent.csv', 'output.csv') def test_missing_columns(): """测试数据列缺失""" # 模拟缺少price列的数据 test_data = """user_id,event_time,product_id,region,device_type 123,1646123456,1001,北京,mobile 124,1646123457,1002,上海,desktop""" df = pd.read_csv(StringIO(test_data)) # 应该能处理缺失列，用NaN填充 assert 'price' in df.columns or True # 根据实际逻辑调整 def test_all_invalid_prices(): """测试所有价格都无效的情况""" test_data = """user_id,event_time,product_id,price,region,device_type 123,1646123456,1001,无效价格,北京,mobile 124,1646123457,1002,也是无效,上海,desktop""" df = pd.read_csv(StringIO(test_data)) df_cleaned = clean_ecommerce_data_from_df(df) # 假设有这个函数 assert df_cleaned['price'].isna().all() def test_timestamp_formats(): """测试各种时间戳格式""" test_cases = [ ('1646123456', '2022-03-01'), # 秒级 ('1646123456123', '2022-03-01'), # 毫秒级 ('2022-03-01 12:00:00', '2022-03-01'), ('01/03/2022', '2022-03-01') # 不同日期格式 ] for input_ts, expected_date in test_cases: result = parse_timestamp(input_ts) assert result.strftime('%Y-%m-%d') == expected_date