使用Sqoop或PySpark导入数据源时如何处理损坏错位的异常数据

你遇到的问题本质是message_text字段含双引号，Sqoop、PySpark默认会将双引号识别为字段包裹符，导致双引号后的分隔符被判定为字段内容，最终出现列偏移。以下按优先级从高到低给出可行方案：

方案1：Sqoop导入阶段修复（最优，从源头解决问题） #

你当前的Sqoop命令没有配置引号相关的转义参数，直接导出时会把字段内的双引号当成字段边界标记，导致行结构损坏。修改Sqoop命令添加引号处理参数即可：

sqoop import --connect jdbc:mysql:{source-connection-string} \
--table SRC_ATM_TRANS \
--username {username} \
--password {password} \
--target-dir /user/root/etl_project \
--fields-terminated-by '|' \
--lines-terminated-by "\n" \
--optionally-enclosed-by '"' \
--escaped-by '\\' \
-m 1

参数说明：

• --optionally-enclosed-by '"'：告诉Sqoop源表中部分字段用双引号包裹，导入时会自动处理字段内的特殊字符
• --escaped-by '\\'：字段内出现的双引号会自动加转义符，避免破坏分隔规则
  另外注意原命令中{username}和--password之间缺了空格，会导致参数解析错误。

方案2：PySpark读取阶段修复（无需重新导入） #

如果已经完成Sqoop导入不想重复跑任务，可以在PySpark读取CSV时关闭引号识别，直接按分隔符拆分所有列：

transactions = spark.read \
.option("sep","|") \
.option("quote", "") # 关闭双引号作为字段包裹符的默认规则
.option("columnNameOfCorruptRecord", "corrupt_col") # 可选，异常行存入单独列方便排查
.csv("/user/root/etl_project/part-m-00000", header = False,schema = transaction_schema)

读取完成后，再对message_text字段单独做清洗：去掉字段首尾多余的双引号、替换字段内不符合规范的分隔符即可。

方案3：已生成错位DataFrame的修复方案 #

如果已经生成了偏移的DataFrame，可以直接对异常行做字段拆分补全：

1. 首先筛选异常行：只要最后n个字段（你的例子里是天气相关的13个字段）全为null的就是异常行
2. 把异常行的message_text字段按|拆分，得到的数组从第2位开始依次补到后面的空字段中
3. 清洗完成后和正常行做合并即可，示例代码：

from pyspark.sql.functions import split, col, when

# 拆分message_text获取后续列的值
split_msg = split(col("message_text"), "\\|")

# 补全所有错位字段
fixed_df = transactions.withColumn("msg_part", split_msg) \
.withColumn("fixed_message_text", col("msg_part")[0]) \
.withColumn("weather_lat", when(col("weather_lat").isNull(), col("msg_part")[1]).otherwise(col("weather_lat"))) \
.withColumn("weather_lon", when(col("weather_lon").isNull(), col("msg_part")[2]).otherwise(col("weather_lon")))
# 后续剩余字段按上述逻辑依次补全，最后删除msg_part、原message_text字段，把fixed_message_text重命名为message_text即可

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