在数据分析项目中，原始数据往往存在重复记录、缺失字段、格式混乱、类型错误等问题。如果全部清洗逻辑都写在MySQL的存储过程或复杂SQL里，执行计划可能变得难以维护，而且某些文本处理或统计建模操作在SQL中表达非常笨拙。反过来，如果把所有数据先全部拉到Julia内存中再清洗，又可能面临内存溢出和网络传输瓶颈。更务实的做法是划分清洗阶段：利用MySQL擅长集合运算和索引扫描的特点，先完成粗粒度的去重、空值删除和条件过滤；再让Julia接手需要逐行逻辑判断、正则替换或分位数检测的细粒度任务。
下面通过一个电商订单表orders来逐步演示。假设该表包含订单号order_id、用户ID user_id、下单时间created_at、支付金额amount、收货地址address和状态status。原始数据中可能存在重复订单、空地址、金额为负的脏数据以及日期格式不一致等问题。
第一阶段：用MySQL完成基础去重与空值过滤
MySQL在处理大规模数据时具有I/O优化和索引加速的优势，适合先执行删除重复行、剔除关键字段为NULL的记录。对于重复订单，可以保留同一订单号下创建时间最早的一条。利用窗口函数ROW_NUMBER()配合公共表表达式CTE即可实现。注意MySQL 8.0以上版本才支持窗口函数，如果还在用5.7，可以使用自连接或临时表加GROUP BY的方式替代。
-- 删除重复订单，保留每个order_id最早的一条
WITH ranked_orders AS (
SELECT *,
ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY created_at ASC) AS rn
FROM orders
)
DELETE FROM orders
WHERE order_id IN (
SELECT order_id FROM ranked_orders WHERE rn > 1
);
上述SQL中，PARTITION BY order_id把相同订单号分为一组，ORDER BY created_at ASC让最早记录排名为1。之后删除排名大于1的记录即可。如果业务要求保留最新一条，改成ORDER BY created_at DESC。但是直接DELETE大量重复数据可能锁表，建议先备份或者分批执行：每次只删除一部分rn大于1的行，比如限制删除1000条，循环直到没有重复。
对于缺失值，要看具体字段的业务意义。收货地址不能为空，否则无法配送；金额不能为NULL或负数。用一条DELETE语句即可完成：
DELETE FROM orders WHERE address IS NULL OR address = '' OR amount IS NULL OR amount < 0;
执行完这两步，orders表中只剩下地址非空、金额合法且订单号唯一的记录。这阶段数据量通常会大幅下降，为后续导入Julia减轻了压力。空字符串''和NULL在MySQL中是不同的，清洗时务必两个条件都检查，否则可能漏掉空字符串的情况。
第二阶段：将数据导入Julia进行细粒度清洗
MySQL擅长集合运算，但处理正则表达式替换、字符串模糊匹配、基于分位数的异常值检测等操作时，写法冗长且性能不高。Julia的DataFrames.jl和StringDistances.jl等包提供了更灵活的逐行处理能力。首先需要把MySQL数据导出，最通用的方式是通过SELECT INTO OUTFILE导出CSV，或者使用ODBC.jl、JDBC连接直接查询返回DataFrame。
推荐使用ODBC.jl连接MySQL，避免频繁的磁盘文件交互。下面代码展示了如何建立连接并读取数据到DataFrame：
using ODBC, DataFrames
# 建立DSN连接，具体DSN名称取决于系统配置
conn = ODBC.Connection("MySQL_DSN", "username", "password")
df = ODBC.query(conn, "SELECT * FROM orders") |> DataFrame
ODBC.close!(conn)
拿到DataFrame后，可以先检查字段类型。describe(df)能显示每个列的类型、缺失值数量和基本统计量。很多CSV导入时金额会被读成字符串，需要转换成数值类型。使用parse.(Float64, df.amount)会报错如果字段中存在非数字字符，所以更稳妥的方式是定义转换函数并处理异常。
对于日期格式不一致的问题，比如created_at字段中既有2023-01-15又有2023/1/15，需要统一成标准格式。常见做法是先用正则提取数字组合，再构造Date对象。Julia的Dates模块提供了灵活的解析方法，配合try-catch可以逐行处理无效日期。
using Dates
function clean_date(dt_str)
# 移除所有非数字字符，只保留年月日
digits_only = replace(dt_str, r"[^0-9]" => "")
if length(digits_only) == 8
y = parse(Int, digits_only[1:4])
m = parse(Int, digits_only[5:6])
d = parse(Int, digits_only[7:8])
return Date(y, m, d)
else
return missing
end
end
df.created_at = [clean_date(s) for s in df.created_at]
这段代码把created_at列中所有值通过列表推导式逐条转换，无法解析的置为missing。注意在Julia中missing表示缺失值，后续分析时需要特别处理。DataFrames.jl支持skipmissing函数过滤掉缺失值。
另一个常见任务是地址标准化。例如地址中可能存在全角空格、多余换行符或邮编位置不统一。可以使用正则表达式清除不可见字符，然后统一地区名称大小写。StringDistances包还能帮助检测地址是否与标准库匹配，但这里先做基础清理即可。
# 清理地址中的多余空白和换行 df.address = [strip(replace(a, r"\s+" => " ")) for a in df.address] # 将地区名称首字母大写 df.address = [titlecase(lowercase(a)) for a in df.address]
通过以上步骤，df中的created_at和address列已经完成规范化。接下来处理异常值。对于金额列，使用分位数法检测离群点。先计算该列的1%和99%分位数，然后把超出范围的记录标记出来。业务上通常不会直接删除，而是单独导出供人工审核。
using Statistics
q1 = quantile(skipmissing(df.amount), 0.01)
q99 = quantile(skipmissing(df.amount), 0.99)
outliers = df[(df.amount .< q1) .| (df.amount .> q99), :]
println("发现 $(nrow(outliers)) 条异常金额记录")
第三阶段：回写MySQL并验证清洗结果
数据在Julia中完成清洗后，需要将结果回写到MySQL，以便后续分析和共享。回写前先确认目标表结构，可以通过CREATE TABLE cleaned_orders LIKE orders复制原表结构，再使用ODBC.jl的插入功能逐批写入。对于较大数据量，逐行INSERT效率很低，推荐使用批量绑定参数或先生成CSV文件再通过LOAD DATA INFILE导入。
使用ODBC的insert!函数可以一次性插入整个DataFrame，但需要注意数据库字段顺序和类型匹配。如果df中列的顺序与表结构不一致，需要先用select或names调整。下面演示先删除目标表旧数据再批量插入的方法：
# 清空目标表
ODBC.execute!(conn, "TRUNCATE TABLE cleaned_orders")
# 将DataFrame写入MySQL
ODBC.load!(conn, df, "cleaned_orders")
# 验证写入行数
nrows_check = ODBC.query(conn, "SELECT COUNT(*) AS cnt FROM cleaned_orders")[1, :cnt]
println("清洗后写入 $nrows_check 行")
如果数据量超过百万行，ODBC.load!可能因为占位符数量限制而失败。这时推荐先通过CSV.write把DataFrame保存为临时CSV，再调用MySQL的LOAD DATA命令。注意CSV文件中的反斜杠和特殊字符需要转义，可在写入时设置escapechar='\\'参数。回写完文件后，在MySQL客户端执行类似下面的语句：
LOAD DATA LOCAL INFILE 'C:/tmp/cleaned_orders.csv' INTO TABLE cleaned_orders FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\n' IGNORE 1 ROWS;
路径中的反斜杠必须保留，Windows系统下文件路径分隔符就是反斜杠。如果使用Linux服务器，路径改为类似/tmp/cleaned_orders.csv。导入完成后，用SQL对比原表和清洗后表的行数、空值数量以及关键字段的统计值，确认清洗没有引入新的问题。
整个清洗流程可以封装成一个Julia脚本，通过调度工具定期执行。每当有新数据进入MySQL，先用SQL做粗清洗，然后触发Julia脚本读取、细清洗并回写，最后发送质量报告。这样既利用了数据库的批量操作性能，又发挥了Julia在复杂逻辑上的表达力，相比单一工具的方案更容易维护和扩展。