ADF管道源端失败：SQL会话被终止问题排查求助

执行6个月约20亿条记录的批量复制任务，使用Azure Data Factory（ADF）Copy Activity从SQL Server读取数据，写入Azure ADLS Gen2存储账户的Parquet文件。已设置管道超时为2天，但持续收到源端失败错误：

"message": "Failure happened on 'Source' side. 'Type=Microsoft.Data.SqlClient.SqlException,Message=111202;Query QID67099467 has been cancelled.\r\nCannot continue the execution because the session is in the kill state.\r\nA severe error occurred on the current command. The results, if any, should be discarded.,Source=Framework Microsoft SqlClient Data Provider,'"

管道JSON配置如下：

{
    "name": "CopyPipeline_abc_6months_load",
    "properties": {
        "activities": [
            {
                "name": "ForEach_TableCopy",
                "type": "ForEach",
                "dependsOn": [],
                "userProperties": [],
                "typeProperties": {
                    "items": {
                        "value": "@pipeline().parameters.tableItems",
                        "type": "Expression"
                    },
                    "activities": [
                        {
                            "name": "Copy_to_parquet",
                            "type": "Copy",
                            "dependsOn": [],
                            "policy": {
                                "timeout": "0.20:00:00",
                                "retry": 3,
                                "retryIntervalInSeconds": 30,
                                "secureOutput": false,
                                "secureInput": false
                            },
                            "userProperties": [],
                            "typeProperties": {
                                "source": {
                                    "type": "SqlServerSource",
                                    "sqlReaderQuery": {
                                        "value": "@concat('select * from ', item().source.table, ' where year([', item().source.dateColumn, '])=2024 and month([', item().source.dateColumn, '])<7 order by month([', item().source.dateColumn, '])')",
                                        "type": "Expression"
                                    },
                                    "partitionOption": "None"
                                },
                                "sink": {
                                    "type": "ParquetSink",
                                    "storeSettings": {
                                        "type": "AzureBlobFSWriteSettings"
                                    },
                                    "formatSettings": {
                                        "type": "ParquetWriteSettings"
                                    }
                                },
                                "enableStaging": false,
                                "validateDataConsistency": true,
                                "translator": {
                                    "type": "TabularTranslator",
                                    "typeConversion": true,
                                    "typeConversionSettings": {
                                        "allowDataTruncation": true,
                                        "treatBooleanAsNumber": false
                                    }
                                }
                            },
                            "inputs": [
                                {
                                    "referenceName": "SourceDataset_b3v",
                                    "type": "DatasetReference",
                                    "parameters": {
                                        "cw_table": "@item().source.table"
                                    }
                                }
                            ],
                            "outputs": [
                                {
                                    "referenceName": "DestinationDataset_b3v",
                                    "type": "DatasetReference",
                                    "parameters": {
                                        "cw_fileName": "@concat(item().destination.filePrefix, '_', formatDateTime(utcnow(), 'yyyyMMddHHmmss'), '.parquet')",
                                        "cw_folder": "@{item().destination.folder}"
                                    }
                                }
                            ]
                        }
                    ]
                }
            }
        ],
        "parameters": {
            "tableItems": {
                "type": "Array",
                "defaultValue": [
                    {
                        "source": {
                            "table": "abc.123",
                            "dateColumn": "date"
                        },
                        "destination": {
                            "filePrefix": "123",
                            "folder": "123_ingest"
                        }
                    },
                    {
                        "source": {
                            "table": "abc.345",
                            "dateColumn": "event_date"
                        },
                        "destination": {
                            "filePrefix": "345",
                            "folder": "345_ingest"
                        }
                    }
                ]
            }
        },
        "folder": {
            "name": "Projects/abc Analytics"
        },
        "annotations": []
    }
}

排查方案 #

1. 确认SQL Server端会话终止原因 #

• 查看SQL Server错误日志（SSMS路径：管理->SQL Server日志），搜索错误码111202和对应QID，确认是资源耗尽（CPU/内存/磁盘IO）还是数据库引擎主动终止（如查询超时、资源调控器限制）
• 执行以下查询，获取会话终止时的资源使用详情：

SELECT 
    req.session_id,
    req.status,
    req.command,
    req.cpu_time,
    req.total_elapsed_time,
    req.logical_reads,
    req.reads,
    req.writes,
    er.message,
    er.error_number
FROM sys.dm_exec_requests req
JOIN sys.dm_exec_sessions ses ON req.session_id = ses.session_id
LEFT JOIN sys.dm_exec_errors er ON req.session_id = er.session_id
WHERE req.session_id = <被终止的会话ID> -- 从错误日志或ADF监控中提取

• 检查资源调控器配置，确认是否对ADF连接账号设置了资源配额

2. 优化ADF源端读取策略 #

• 启用动态分区读取：当前partitionOption为None，针对日期列拆分大查询为小批次，降低单会话负载：
  修改SqlServerSource配置：

  "source": {
      "type": "SqlServerSource",
      "sqlReaderQuery": {
          "value": "@concat('select * from ', item().source.table, ' where [', item().source.dateColumn, '] >= ''2024-01-01'' and [', item().source.dateColumn, '] < ''2024-07-01''')",
          "type": "Expression"
      },
      "partitionOption": "DynamicRange",
      "partitionSettings": {
          "partitionColumnName": "@item().source.dateColumn",
          "partitionLowerBound": "2024-01-01",
          "partitionUpperBound": "2024-07-01",
          "partitionCount": 6
      }
  }
  

• 移除不必要排序：删除查询中的order by month([dateColumn])，ADF写入Parquet无需源端排序，减少SQL Server排序开销
• 限制并发连接数：在SqlServerSource中添加maxConcurrentConnections参数（如设为5），避免并发过高压垮源端

3. 优化SQL查询性能 #

• 为日期列创建非聚集索引，避免全表扫描：

CREATE NONCLUSTERED INDEX IX_<TableName>_<DateColumn> 
ON <TableName> (<DateColumn>)
INCLUDE (<需复制的列名列表>) -- 若为select *，可创建覆盖索引

• 改用范围查询替代函数过滤：将year([dateColumn])=2024 and month([dateColumn])<7改为[dateColumn] >= '2024-01-01' and [dateColumn] < '2024-07-01'，让SQL Server能利用日期索引

4. 调整ADF管道执行策略 #

• 控制ForEach并行度：将ForEach设置为串行执行，或降低并发数：

"typeProperties": {
    "items": {
        "value": "@pipeline().parameters.tableItems",
        "type": "Expression"
    },
    "isSequential": true,
    "activities": [ ... ]
}

• 启用暂存机制：设置enableStaging: true，用Azure Blob Storage作为暂存层，先导出数据到暂存再加载到ADLS Gen2，分散源端负载

5. 实时监控定位问题 #

• 在ADF监控面板查看每个Copy Activity的执行时长、数据读取速率，确认是否是特定表的复制导致会话终止
• 启用SQL Server Query Store，跟踪ADF执行的SQL语句的资源消耗和执行计划，定位性能瓶颈

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