pipeline-migration

遥测 — 强制要求。 每次 api.fabric.microsoft.com 调用都必须携带 x-ms-fabric-skill: pipeline-migration（az rest：--headers "x-ms-fabric-skill=pipeline-migration"）， 包括每次 LRO 轮询、fabric_lro 和重试。代码片段中省略了该项——仍需添加。

重要说明

1. 若要根据工作区名称查找工作区详细信息（包括其 ID）：列出所有工作区，然后使用 JMESPath 筛选
2. 若要根据工作区 ID、项类型和项名称查找项详细信息（包括其 ID）：列出该工作区中该类型的所有项，然后使用 JMESPath 筛选
3. Fabric 没有 “Dataset” 项类型——所有数据集属性都内联在活动的 typeProperties 中
4. 链接服务映射为 Fabric 连接——在管道活动 JSON 中，活动 linkedService 块上的 referenceName 使用 Fabric 连接的显示名称（而非连接 GUID）；连接 GUID 仅用于 Fabric REST API 调用
5. Notebook 活动从 SynapseNotebook 改为 TridentNotebook，并通过 GUID（而非名称）引用 notebook
6. Synapse 全局参数在 Fabric 中变为变量库项，并通过 @pipeline().libraryVariables.<name> 引用。变量库的 Number 类型无法在管道中使用——为确保运行时兼容性，Synapse Float/Double 会映射为 String
7. Fabric 中不存在 Validation 活动类型——必须将其改写为 GetMetadata + IfCondition
8. 触发器有意排除在此技能之外——迁移后，请在 Fabric 中手动重新创建计划
9. SSIS 包执行、仅 SHIR 支持的连接器和 Databricks 活动已暂缓处理——请参阅 pipeline-gotchas.md

Synapse Pipelines → Microsoft Fabric Data Factory 迁移

前置知识

以下配套文档介绍了通用的 Fabric REST 模式。请勿预先阅读——仅在某个特定阶段需要此技能资源文件尚未涵盖的模式时参考：

• COMMON-CORE.md — 通用 Fabric REST API 模式、身份验证与令牌受众、通过 JMESPath 发现项
• COMMON-CLI.md — az rest / az login 命令行模式、身份验证方法、管道运行/计划操作
• ITEM-DEFINITIONS-CORE.md — DataPipeline 和 VariableLibrary 项定义结构（pipeline-content.json、variables.json）
• SPARK-AUTHORING-CORE.md — 创建 Fabric notebook 项（Fabric 中尚无 notebook 项时需要）

关于迁移的 notebook 部分（即管道活动调用的 notebook），请使用配套的 synapse-migration 技能迁移 notebook 内容本身。

---

目录

主题	参考
迁移前评估（首先运行）	pipeline-assessment.md
迁移编排器	pipeline-orchestrator.md
API 驱动的迁移工作流	§ API 驱动的迁移工作流
活动类型映射	activity-mapping.md
Notebook 活动迁移（主要关注点）	notebook-activity-migration.md
链接服务 → 连接	linked-service-to-connection.md
数据集内联	dataset-inlining.md
全局参数 → 变量库	global-parameters-to-variable-library.md
管道注意事项与暂缓处理的活动	pipeline-gotchas.md
验证与测试	validation-testing.md
迁移报告	migration-report.md

上下文加载指南

**重要——仅加载所需内容。**请勿预先读取所有资源文件。

何时	读取此文件
用户在迁移之前要求评估、范围界定或计划	pipeline-assessment.md
用户要求迁移整个管道工作区	pipeline-orchestrator.md
用户询问活动类型映射或不受支持的活动	activity-mapping.md
用户有 SynapseNotebook 活动（最常见）	notebook-activity-migration.md
用户询问链接服务或连接	linked-service-to-connection.md
用户询问使用数据集的 Copy、Lookup 或 GetMetadata	dataset-inlining.md
用户需要转换全局参数	global-parameters-to-variable-library.md
用户遇到 SSIS、SHIR、Databricks 或其他阻碍	pipeline-gotchas.md
迁移后验证	validation-testing.md
生成迁移摘要	migration-report.md

---

API 驱动的迁移工作流

身份验证

目标	令牌受众
Synapse 数据平面（管道、数据集、链接服务）	https://dev.azuresynapse.net
Synapse ARM（全局参数、工作区属性）	https://management.azure.com
Fabric REST API（创建管道、连接、Variable Libraries）	https://api.fabric.microsoft.com

使用 az account get-access-token --resource <audience> --query accessToken -o tsv 获取令牌。

Synapse 数据平面 API 参考

操作	终结点
列出所有管道	GET https://{ws}.dev.azuresynapse.net/pipelines?api-version=2020-12-01
获取管道定义	GET https://{ws}.dev.azuresynapse.net/pipelines/{name}?api-version=2020-12-01
列出所有数据集	GET https://{ws}.dev.azuresynapse.net/datasets?api-version=2020-12-01
获取数据集定义	GET https://{ws}.dev.azuresynapse.net/datasets/{name}?api-version=2020-12-01
列出所有链接服务	GET https://{ws}.dev.azuresynapse.net/linkedservices?api-version=2020-12-01
获取链接服务	GET https://{ws}.dev.azuresynapse.net/linkedservices/{name}?api-version=2020-12-01
获取工作区（全局参数）	GET https://management.azure.com/subscriptions/{subId}/resourceGroups/{rg}/providers/Microsoft.Synapse/workspaces/{ws}?api-version=2021-06-01

Fabric API 参考

操作	终结点
列出连接	GET https://api.fabric.microsoft.com/v1/connections
创建连接	POST https://api.fabric.microsoft.com/v1/connections
创建管道项	POST https://api.fabric.microsoft.com/v1/workspaces/{wsId}/items
更新管道定义	POST https://api.fabric.microsoft.com/v1/workspaces/{wsId}/items/{id}/updateDefinition
获取管道定义	POST https://api.fabric.microsoft.com/v1/workspaces/{wsId}/items/{id}/getDefinition
创建 Variable Library	POST https://api.fabric.microsoft.com/v1/workspaces/{wsId}/items（类型：VariableLibrary）
列出工作区中的笔记本	GET https://api.fabric.microsoft.com/v1/workspaces/{wsId}/notebooks

---

评估模式（可选，但建议使用）

在 Fabric 中创建任何项目之前，请先运行管道评估，以了解范围、复杂性和阻碍因素。评估为只读操作——仅查询 Synapse API，并生成 Markdown 报告。

Copilot 工作流——无需 Python 脚本文件：

1. 询问用户：“你的 Synapse 工作区名称是什么？”
2. 通过 az account show 和 az synapse workspace show 自动发现订阅 ID 和资源组
3. 在终端中直接运行评估代码
4. 在聊天中直接输出完整报告

使用时机	操作
用户想在决定迁移前了解迁移范围	询问工作区名称 → 加载 pipeline-assessment.md → 直接运行
用户询问“哪些内容会迁移，哪些不会？”	运行评估，并展示“执行摘要”部分
用户请求迁移计划或范围界定文档	运行评估，并在聊天中输出报告
用户已经决定迁移	跳过评估——直接进入下方的迁移阶段

若要同时将报告保存到磁盘，请向 generate_assessment_report() 传入 output_path=f"pipeline-assessment-{SYNAPSE_WS}.md"。生成的 PipelineAssessment 对象可直接提供给 pipeline-orchestrator.md 中的迁移脚本使用。

---

迁移模式（内联执行——无需脚本文件）

Copilot 直接在终端中执行迁移。无需保存或手动运行 Python 文件。

向用户询问：

1. Synapse 工作区名称（如果已运行评估，则沿用之前的名称）
2. Fabric 工作区名称
3. 要迁移的管道——指定名称，或使用 * 表示全部
4. 要追加到 Fabric 中每个管道名称的可选后缀（例如 _migrated）——留空则保留原名称

其他信息均会自动发现：

• 通过 az account show + az synapse workspace show 获取订阅 ID 和资源组
• 通过按显示名称筛选的 GET /v1/workspaces 获取 Fabric 工作区 ID
• 通过 GET /v1/workspaces/{wsId}/notebooks 获取 Fabric 中的 Notebook GUID（SynapseNotebook → TridentNotebook 所必需）
• 通过 GET /v1/connections 获取 Fabric 中的连接名称（当数据集引用链接服务时）

内联自动完成的操作：

• ✅ SynapseNotebook → TridentNotebook——重命名类型、查找 GUID、移除 sparkPool/sessionConfiguration，并将超时修正为最长 12h
• ✅ 所有兼容的活动类型——直接传递，仅调整少量属性
• ✅ 将数据集内联到活动的 typeProperties 中
• ✅ 将全局参数表达式重写为 @pipeline().libraryVariables.<name>
• ✅ 组装管道 JSON，并通过 REST 部署到 Fabric

开始前，Copilot 会进行检查——遇到以下情况时会暂停并报告：

• SynapseNotebook 活动引用的 Notebook 尚未存在于 Fabric 工作区中
• 数据集活动引用的链接服务缺少对应的 Fabric 连接

加载 pipeline-orchestrator.md 以获取完整的内联运行程序。

---

迁移阶段（按顺序执行）

阶段	来源	目标	资源
阶段 0	Synapse 笔记本（由管道活动引用）	Fabric Notebooks	synapse-migration 技能
阶段 1	Synapse 全局参数	Fabric Variable Library	global-parameters-to-variable-library.md
阶段 2	Synapse 链接服务	Fabric Connections	linked-service-to-connection.md
阶段 3	Synapse 数据集	内联到活动中	dataset-inlining.md
阶段 4	Synapse 管道活动	Fabric 管道活动	activity-mapping.md + notebook-activity-migration.md
阶段 5	组装后的管道 JSON	Fabric DataPipeline 项	pipeline-orchestrator.md
最终	—	验证	validation-testing.md

阶段 0 必须先于阶段 4：编写 TridentNotebook 活动前，需要先获取 Fabric Notebook GUID。 阶段 2 必须先于阶段 3：在内联数据集中引用连接名称前，必须先创建这些连接。

---

活动类型速查

完整映射表、迁移前后的示例以及暂缓处理的决策，参见 activity-mapping.md。

Synapse 活动	Fabric 对应项	状态
SynapseNotebook	TridentNotebook	✅ 已迁移 — 参见 notebook-activity-migration.md
Copy	Copy	✅ 已迁移 — 数据集已内联
Lookup	Lookup	✅ 已迁移 — 数据集已内联
GetMetadata	GetMetadata	✅ 已迁移 — 数据集已内联
Validation	GetMetadata + IfCondition	✅ 已迁移 — 拆分为 2 个活动
ForEach	ForEach	✅ 兼容
IfCondition	IfCondition	✅ 兼容
Switch	Switch	✅ 兼容
Until	Until	✅ 兼容
Wait	Wait	✅ 兼容
Fail	Fail	✅ 兼容
SetVariable	SetVariable	✅ 兼容
AppendVariable	AppendVariable	✅ 兼容
ExecutePipeline	ExecutePipeline	✅ 兼容 — 添加 workspaceId
WebActivity	WebActivity	✅ 兼容
Script	Script	✅ 兼容 — 更新连接引用
Delete	Delete	✅ 已迁移 — 数据集已内联
Filter	Filter	✅ 兼容
SparkJobDefinition（Synapse SJD）	SparkJobDefinition	✅ 更新 GUID 引用
HDInsightSpark	TridentNotebook 或 SparkJobDefinition	⚠️ 需要重写
AzureMLBatchExecution	WebActivity	⚠️ 重写为 REST 调用
AzureFunctionActivity	WebActivity	⚠️ 需要重写 — 使用函数 URL 和密钥
DatabricksNotebook	⛔ 暂缓处理	参见 pipeline-gotchas.md
DatabricksSparkJar	⛔ 暂缓处理	参见 pipeline-gotchas.md
DatabricksSparkPython	⛔ 暂缓处理	参见 pipeline-gotchas.md
ExecuteSSISPackage	⛔ 暂缓处理	参见 pipeline-gotchas.md
AzureBatch	⛔ 暂缓处理	Fabric 中没有对应项
Custom	⛔ 暂缓处理	Fabric 中没有对应项

---

必须 / 优先 / 避免

必须执行

• 先迁移笔记本，再迁移管道 — Fabric TridentNotebook 活动需要笔记本 GUID，而不是名称。首先使用 synapse-migration 技能
• 先创建 Fabric 连接，再构建管道 JSON — Synapse 中的链接服务名称在 Fabric 中会成为连接引用；内联数据集之前，需要先获取连接名称
• 内联所有数据集定义 — Fabric Data Factory 没有 Dataset 项类型；每个活动的所有 inputs/outputs 数据集属性都必须嵌入其中
• 创建 Variable Library 后，将 @pipeline().globalParameters.<name> 替换为 @pipeline().libraryVariables.<name>
• 将 Validation 活动 替换为 GetMetadata + IfCondition 组合 — Fabric 中不存在 Validation 活动类型
• 从迁移后的笔记本活动中移除 sparkPool 和 sessionConfiguration — 池选择和会话配置应放在笔记本关联的 Fabric Environment 中

建议

• 为 dev/test/prod 环境使用带 Value Sets 的 Variable Library — 使用带环境专属 Value Sets 的 @pipeline().libraryVariables.<name>，而不是管道级参数来提升环境
• 在 TridentNotebook 活动中优先使用参数化的 notebookParameters，而不是硬编码值 — 与 Synapse 参数化笔记本模式一致
• 运行完整迁移管道前，逐个测试笔记本活动 — 笔记本 GUID 是最常见的故障来源
• 在适用的 Copy 活动中使用 OneLake Lakehouse 源/接收器 — 对于已在 OneLake 中的数据，无需外部连接

避免

• 不要在 TridentNotebook 活动中通过名称引用笔记本 — 使用 Fabric 工作区笔记本列表中的 GUID
• 不要沿用 Synapse 触发器 — 此技能不会迁移这些触发器；验证管道后，请在 Fabric 中重新创建计划
• 未先阅读 pipeline-gotchas.md 前，不要尝试迁移 SSIS、Databricks 或 AzureBatch 活动 — 这些活动需要手动处理
• 不要在管道 JSON 中硬编码工作区/项目 GUID — 使用 Variable Library 条目或管道参数，以便无需编辑管道 JSON 即可提升环境
• 迁移后不要使用 @pipeline().globalParameters 语法 — Fabric 中不存在此表达式路径；所有迁移的全局参数都必须通过 @pipeline().libraryVariables 访问

---

迁移注意事项 — 快速参考

完整的故障排除指南请参阅 pipeline-gotchas.md。

#	Flag ID	问题	严重性	解决方案摘要
PG1	NOTEBOOK_GUID_NOT_FOUND	构建 TridentNotebook 活动时，Notebook 尚未迁移到 Fabric	高	先运行 synapse-migration 技能；通过 GET /v1/workspaces/{wsId}/notebooks 获取 GUID
PG2	DATASET_NOT_INLINED	活动仍引用命名数据集（Fabric 不支持）	高	应用 dataset-inlining.md 中的模式，将数据集属性嵌入活动源或接收器
PG3	GLOBAL_PARAM_EXPRESSION	迁移后的管道中仍有 @pipeline().globalParameters.<name> 表达式	高	创建 Variable Library 后，将其替换为 @pipeline().libraryVariables.<name>
PG4	VALIDATION_ACTIVITY_UNSUPPORTED	管道 JSON 中仍有 Validation 活动类型	高	改写为 GetMetadata + IfCondition — 参阅 activity-mapping.md
PG5	SHIR_CONNECTOR_PARKED	活动使用由 Self-Hosted Integration Runtime 支持的链接服务	中	必须在 Fabric 中设置本地数据网关；参阅 pipeline-gotchas.md
PG6	SSIS_ACTIVITY_PARKED	ExecuteSSISPackage 活动无法迁移	高	暂缓处理 — Fabric 中没有对应项；替代方案请参阅 pipeline-gotchas.md
PG7	DATABRICKS_ACTIVITY_PARKED	Databricks 活动类型没有 Fabric 原生对应项	高	暂缓处理 — 可通过 WebActivity 使用 Databricks REST API 作为变通方案；参阅 pipeline-gotchas.md
PG8	SPARKPOOL_REF_ORPHANED	TridentNotebook 活动中仍有 sparkPool / targetBigDataPool 引用	中	删除 sparkPool 和 sessionConfiguration 块；池配置应放在 Fabric Environment 中
PG9	EXECUTE_PIPELINE_NO_WORKSPACE	ExecutePipeline 活动缺少所引用管道的 workspaceId	中	将 workspaceId 添加到 typeProperties；即使是同一工作区的子管道也必须提供 — 省略会导致运行时故障
PG10	LINKED_SERVICE_NO_CONNECTION	链接服务没有匹配的 Fabric Connection	高	手动或通过 API 创建 Connection；更新内联数据集中的 Connection 引用

---

迁移后：接下来做什么

管道迁移完成后，将后续工作交接给以下配套技能和工具：

任务	技能 / 工具
迁移 Notebook 内容（mssparkutils → notebookutils、链接服务）	synapse-migration 技能
为已迁移的管道安排计划	COMMON-CLI.md § 作业计划
监控管道运行	Fabric 工作区 → Monitor 中心
构建新的 Fabric 管道	参阅 ITEM-DEFINITIONS-CORE.md § DataPipeline
管道运行后浏览已迁移的 Lakehouse 数据	spark-cli 或 sqldw-cli 技能

---

示例

SynapseNotebook → TridentNotebook 活动（迁移前/后）

{
  "name": "Run_Notebook",
  "type": "SynapseNotebook",
  "typeProperties": {
    "notebook": {"referenceName": "MyNotebook", "type": "NotebookReference"},
    "sparkPool": {"referenceName": "BigPool", "type": "BigDataPoolReference"}
  }
}

迁移后（GUID 通过 GET /v1/workspaces/{wsId}/notebooks 获取）：

{
  "name": "Run_Notebook",
  "type": "TridentNotebook",
  "typeProperties": {
    "notebookId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    "workspaceId": "yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy"
  }
}

全局参数表达式改写

Before (Synapse):  @pipeline().globalParameters.batchDate
After (Fabric):    @pipeline().libraryVariables.batchDate

数据集内联：Copy 活动 — 数据集引用 → Connection 显示名称

Synapse 数据集 InputBlobDataset（将被内联并移除）：

{
  "type": "AzureBlob",
  "linkedServiceName": {"referenceName": "AzureBlobLinkedService"},
  "typeProperties": {"folderPath": "input/", "fileName": "data.csv"}
}

Synapse Copy 活动（迁移前 — 按名称引用数据集）：

{
  "name": "CopyData", "type": "Copy",
  "inputs": [{"referenceName": "InputBlobDataset", "type": "DatasetReference"}],
  "typeProperties": {"source": {"type": "BlobSource"}, "sink": {"type": "BlobSink"}}
}

迁移后（AzureBlobLinkedService 的 Connection 显示名称为 My ADLS Connection）：

{
  "name": "CopyData", "type": "Copy",
  "typeProperties": {
    "source": {"type": "BlobSource", "storeSettings": {"type": "AzureBlobStorageReadSettings"}},
    "sink":   {"type": "BlobSink",   "storeSettings": {"type": "AzureBlobStorageWriteSettings"}}
  },
  "linkedService": {"referenceName": "My ADLS Connection", "type": "LinkedServiceReference"}
}

完整的迁移前/后示例请参阅 activity-mapping.md 和 notebook-activity-migration.md。