如何在Azure Data Factory中处理HTTP脏CSV数据并映射至Azure SQL

方案1：预处理CSV数据（推荐） #

• 用脚本统一清洗数据格式，确保每行都能拆分为3列：
  1. 读取原始CSV时，启用带转义的双引号解析规则，把带引号的行正确拆分。
  2. 输出标准CSV，保证所有行的列数一致，内部特殊字符处理规范。
     示例Python代码片段：

  import csv
  
  with open('input.csv', 'r', encoding='utf-8') as infile, open('cleaned.csv', 'w', newline='', encoding='utf-8') as outfile:
      reader = csv.reader(infile, quoting=csv.QUOTE_ALL, escapechar='"')
      writer = csv.writer(outfile, quoting=csv.QUOTE_MINIMAL)
      for row in reader:
          # 确保每行固定3列，缺失列补空值
          while len(row) < 3:
              row.append('')
          writer.writerow(row[:3])
  

方案2：Azure Data Factory（ADF）高级配置 #

• 在ADF复制活动的CSV数据集里设置：
  • 引号字符设为"，转义字符设为"
  • 列分隔符保持为,，开启跳过首行（如果有表头）
  • 手动指定架构映射：强制源列对应目标表的3个字段（Col_A、Col_B、Col_B），若ADF自动检测列数错误，手动调整源列数量为3。

方案3：临时表中转导入 #

• 先创建单列临时表导入所有原始行：

  CREATE TABLE #TempCSV (RawData NVARCHAR(MAX))
  BULK INSERT #TempCSV
  FROM 'https://your-http-site/data.csv'
  WITH (
      FIELDTERMINATOR = '\n',
      ROWTERMINATOR = '\n',
      FIRSTROW = 2 -- 跳过表头行
  )
  

• 用T-SQL字符串函数拆分每行，处理带引号和内部逗号的情况：

  INSERT INTO TargetTable (Col_A, Col_B, Col_B)
  SELECT
      -- 提取第一列
      CASE 
          WHEN RawData LIKE '"%' THEN SUBSTRING(RawData, 2, CHARINDEX(',', RawData, 2)-2)
          ELSE SUBSTRING(RawData, 1, CHARINDEX(',', RawData)-1)
      END AS Col_A,
      -- 提取第二列
      CASE
          WHEN RawData LIKE '"%' THEN SUBSTRING(RawData, CHARINDEX(',', RawData, 2)+1, CHARINDEX(',', RawData, CHARINDEX(',', RawData, 2)+1) - CHARINDEX(',', RawData, 2) -1)
          ELSE SUBSTRING(RawData, CHARINDEX(',', RawData)+1, CHARINDEX(',', RawData, CHARINDEX(',', RawData)+1) - CHARINDEX(',', RawData) -1)
      END AS Col_B,
      -- 提取第三列，替换转义引号
      CASE
          WHEN RawData LIKE '%""%' THEN REPLACE(SUBSTRING(RawData, CHARINDEX(',', RawData, CHARINDEX(',', RawData, 2)+1)+1, LEN(RawData)-CHARINDEX(',', RawData, CHARINDEX(',', RawData, 2)+1)-1), '""', '"')
          ELSE SUBSTRING(RawData, CHARINDEX(',', RawData, CHARINDEX(',', RawData)+1)+1, LEN(RawData))
      END AS Col_B
  FROM #TempCSV
  

• 清理临时表：DROP TABLE #TempCSV

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