关于 spark-authoring-cli
spark-authoring-cli 为 Microsoft Fabric 数据工程师提供通过 REST API 管理 Spark 项目及编写 Notebook 代码的技能。内容涵盖工作区、Lakehouse、Notebook 与作业等操作，并提供 PySpark、Scala、SparkR 和 SQL 编写指引。
主要功能
- 01通过 REST API 管理 Fabric 项目
- 02编写 PySpark、Scala、SparkR 和 SQL
- 03指导 Notebook 内容读取与更新
- 04涵盖作业执行和调度
- 05提供身份验证与数据访问指引
适用场景
- 01管理工作区中的 Lakehouse 和 Notebook
- 02执行或调度 Spark 作业
- 03通过 API 修改 Notebook 内容
- 04编写及调试 Fabric Notebook 代码
安装
复制下面的命令，在终端中运行。
$
npx skills add https:/安装后在 Claude Code 中输入/spark-authoring-cli确认已加载。
包含的文件
除 SKILL.md 外，安装时一起下载的参考资料和脚本。
resources/data-engineering-patterns.mddevelopment-workflow.mdinfrastructure-orchestration.mdmaterialized-lake-view-patterns.mdmlv-incremental-refresh-patterns.mdnotebook-api-operations.md详细介绍
遥测 — 强制要求。 每次调用
api.fabric.microsoft.com时都必须携带x-ms-fabric-skill: spark-authoring-cli（az rest：--headers "x-ms-fabric-skill=spark-authoring-cli"）， 包括每次 LRO 轮询、fabric_lro和重试。代码片段中省略了此标头 — 仍须添加。
更新检查 — 每个会话仅一次（强制要求） 在会话中首次使用此技能时，必须先运行 check-updates 技能，再继续操作。
- GitHub Copilot CLI / VS Code：调用
check-updates技能。- Claude Code / Cowork / Cursor / Windsurf / Codex：比较本地与远程的 package.json 版本。
- 如果本会话中较早时已执行过检查，则跳过。
重要说明
- 要通过工作区名称查找工作区详细信息（包括其 ID）：先列出所有工作区，然后使用 JMESPath 进行筛选
- 要通过工作区 ID、项类型和项名称查找项的详细信息（包括其 ID）：先列出该工作区中此类型的所有项，然后使用 JMESPath 进行筛选
Spark 创作 — CLI 技能
此技能涵盖两个互补领域：(1) 通过 REST API 管理 Fabric Spark 项目（工作区、湖屋、笔记本、作业、管道）；(2) 在 Fabric 笔记本单元格中编写代码（PySpark、Scala、SparkR、SQL，以及正确访问湖屋、使用 notebookutils 和配置 Spark）。有关笔记本代码创作基础和共享模块，必须参阅 SPARK-NOTEBOOK-AUTHORING-CORE.md。
目录
必须／优先／避免
必须做到
- 创建新的笔记本运行之前，必须检查近期作业 — 查询过去 5 分钟内的作业实例；如果存在近期作业，应监控该作业，而不是创建重复作业
- POST 后立即记录作业实例 ID — 在执行任何其他操作前保存作业 ID，以便正确监控
- 操作前验证工作区容量分配 — 工作区必须已分配容量且容量处于活动状态
- 用户提供公共数据 URL 时，遵循 Public URL Data Ingestion policy — 将详细行为保留在链接的资源部分，以避免规则不一致或重复
- 正确设置笔记本单元格格式 — 单元格源代码数组中的每一行都必须以
\n结尾，以防代码合并 - 使用正确的 Lakehouse Livy 会话请求正文格式 — 发送包含
name、driverMemory、driverCores、executorMemory、executorCores的扁平 JSON。不要使用{"payload": ...}包装，也不要只发送{"kind": "pyspark"}，否则会导致 HTTP 500。使用有效的内存值（28g、56g、112g、224g）。请参阅下方的 Create Lakehouse Livy Session 示例和 SPARK-CONSUMPTION-CORE.md。
优先做到
- 以适当的间隔轮询作业状态 — 每次轮询间隔 10-30 秒；在合理时长后超时（例如 30 分钟）
- POST 响应不可读时检查作业历史记录 — 如果 POST 返回“无内容”或响应不可读，请在重试前查询近期作业（过去 1 分钟内）
- 开发时使用 Starter Pool — 开发／测试工作负载应使用
useStarterPool: true - 生产环境使用 Workspace Pool — 生产工作负载需要稳定的性能，应使用
useWorkspacePool: true - 创建期间启用湖仓架构 — 设置
creationPayload.enableSchemas: true，以便更好地组织表 - 实现幂等性检查 — 先检查现有状态，防止重复操作
避免
- 绝不要使用相同参数重试 POST — 如果已有作业 ID，只使用 GET 检查状态；不要创建重复的作业实例
- 不要跳过容量验证 — 如果工作区容量已暂停或未分配，操作将会失败
- 避免失败后立即重试 POST — 先检查是否存在现有／活动作业，以防止重复
- 监控现有作业时不要创建新的运行 — 一次只运行一个作业；等待完成后再提交新的运行
- 不要对工作区／湖仓 ID 进行硬编码 — 通过项列表或目录搜索 API 动态发现
- 此处应自行处理
%%sql笔记本单元格，不要转交给sqldw-consumption-cli— “编写一个%%sql单元格”（或任何笔记本魔法命令单元格）的请求属于笔记本单元格编写，即使该单元格查询的是湖仓表。只有当用户需要针对 SQL 终结点执行普通 T-SQL 查询，而非编写笔记本单元格时，才转交给sqldw-consumption-cli。 - 此处应自行处理“运行／执行名为 X 的笔记本” — 按名称运行 Fabric 笔记本属于通过 Jobs API（
RunNotebook）执行笔记本，归属此技能；不要转交给spark-consumption-cli（该技能仅用于执行临时 Livy 会话代码）。 - 不要使用 Lakehouse Livy 会话运行 Fabric 笔记本 — Lakehouse Livy 会话（公共 Livy API）用于临时交互式 Spark 代码执行。要将笔记本作为作业运行，请使用 Jobs API（
RunNotebook），该 API 会在内部创建 Notebook Spark 会话。请参阅 SPARK-AUTHORING-CORE.md § Notebook Execution & Job Management - 不要从笔记本中安排 MLV 刷新 — 如果用户要求“安排 MLV 刷新”，请转交给 mlv-operations-cli，它使用 REST API。基于笔记本的
REFRESH MATERIALIZED LAKE VIEW ... FULL仅用于一次性手动刷新，而非定期计划。
规则 — 请先阅读，并始终遵守
规则 1 — 操作前验证先决条件。 确认工作区已分配容量（参见 COMMON-CORE.md 中的 Create Workspace and Capacity Management），并确认资源 ID 存在，然后再尝试操作。
规则 2 — 信任 updateDefinition 的成功结果。
updateDefinition的轮询结果为Succeeded，足以确认内容和 lakehouse 绑定已持久化。每次上传后都不要调用getDefinition——这是一个会显著增加延迟的异步 LRO。仅在其预期用途下使用getDefinition：在修改前读取当前笔记本内容。规则 3 — 防止重复作业并正确监控执行。 提交新的笔记本运行前，务必先检查近期作业实例（最近 5 分钟）。如果存在近期作业实例，请监控该实例，而不是创建重复作业。提交后，立即记录作业实例 ID 并轮询状态——绝不要重试 POST。请参阅 SPARK-AUTHORING-CORE.md 中的 Job Monitoring，了解相关模式。
规则 4 — 编写笔记本代码时，必须遵循 SPARK-NOTEBOOK-AUTHORING-CORE.md。 在笔记本单元格中编写代码时，必须先阅读 SPARK-NOTEBOOK-AUTHORING-CORE.md——其中定义了代码生成方法、规则，以及链接到详细指南的 Module Index（包括 lakehouse 路径、连接、上下文、编排等）。使用此技能中的 Spark 专用资源（data-engineering-patterns.md、development-workflow.md）了解 Spark 专用实现细节。若任务涉及 Materialized Lake Views，请阅读 materialized-lake-view-patterns.md 了解编写和设计指南，并阅读 mlv-incremental-refresh-patterns.md 进行刷新就绪情况分析。
笔记本代码生成快速规则
常见笔记本编写任务的快速参考。共享的 common/notebook-authoring/ 核心规范（参见规则 4 / SPARK-NOTEBOOK-AUTHORING-CORE.md）具有权威性；如果两者存在差异，请遵循共享核心规范。
| 用户请求 | 必需的输出模式 |
|---|---|
%%sql / 跨 lakehouse 查询单元格 | 返回 Fabric 笔记本 %%sql 单元格。在代码或说明中包含指定的工作区/lakehouse/架构/表。这是笔记本编写，不是交互式 Spark 使用。 |
| 检测管道上下文 | 使用 notebookutils.runtime.context；包含 isForPipeline = context["isForPipeline"]，并读取 context["currentWorkspaceId"]。 |
| 使用 Spark 处理内置笔记本资源文件 | 使用 notebookutils.nbResPath、builtin/ 和 file: 前缀：spark.read.json(f"file:{notebookutils.nbResPath}/builtin/config.json")。Spark 和 notebookutils.fs 在使用资源文件夹的本地路径时需要 file:。 |
| Fabric 连接 | 直接使用 notebookutils.connections.getCredential("{connectionId}")，并展示简洁的代码示例。不要进行网页搜索；将回答控制在工具/轮次预算内。 |
最简代码片段
# Pipeline context
context = notebookutils.runtime.context
isForPipeline = context["isForPipeline"]
workspace_id = context["currentWorkspaceId"]
print(f"isForPipeline={isForPipeline}, workspace_id={workspace_id}")
# Spark read from notebook built-in resources
config_df = spark.read.option("multiline", "true").json(
f"file:{notebookutils.nbResPath}/builtin/config.json"
)
display(config_df)
# PostgreSQL through a Fabric connection (fill in connectionId and host/db)
import psycopg2
credential = notebookutils.connections.getCredential("{connectionId}")
conn = psycopg2.connect(
host="<postgres-host>",
database="<database>",
user=credential["username"],
password=credential["password"],
)
快速入门示例
有关详细模式、身份验证和完整 API 用法，请参阅：
- COMMON-CORE.md — Fabric REST API 模式、身份验证、项目发现
- COMMON-CLI.md —
az rest用法、环境检测、令牌获取 - SPARK-AUTHORING-CORE.md — 笔记本部署、lakehouse 创建、作业执行
以下是最简快速入门示例。生产环境中务必参考 COMMON- 文件。*
创建工作区和 Lakehouse
# See COMMON-CORE.md Environment URLs and SPARK-AUTHORING-CORE.md for full patterns
cat > /tmp/body.json << 'EOF'
{"displayName": "DataEng-Dev"}
EOF
workspace_id=$(az rest --method post --resource "https://api.fabric.microsoft.com" \
--url "https://api.fabric.microsoft.com/v1/workspaces" \
--body @/tmp/body.json --query "id" --output tsv)
cat > /tmp/body.json << 'EOF'
{"displayName": "DevLakehouse", "type": "Lakehouse", "creationPayload": {"enableSchemas": true}}
EOF
lakehouse_id=$(az rest --method post --resource "https://api.fabric.microsoft.com" \
--url "https://api.fabric.microsoft.com/v1/workspaces/$workspace_id/items" \
--body @/tmp/body.json --query "id" --output tsv)
使用架构组织 Lakehouse 表
# See SPARK-AUTHORING-CORE.md Lakehouse Schema Organization for table organization patterns
#
# IMPORTANT: `CREATE SCHEMA` is only supported on schema-enabled lakehouses. On a
# lakehouse created WITHOUT `enableSchemas`, it fails with
# "Feature not supported on Apache Spark in Microsoft Fabric". Always detect schema
# support first, then branch: create schemas only when enabled, otherwise skip.
# Schema names are configurable. If your workspace standard is `raw`/`curated`
# instead of `bronze`/`silver`/`gold`, substitute those names consistently.
import uuid
def lakehouse_is_schema_enabled():
# Probe schema support by attempting CREATE SCHEMA and catching the
# platform-blocked error. Use a unique probe schema name to avoid
# accidentally dropping a real pre-existing schema.
probe = f"_schema_probe_{uuid.uuid4().hex}"
try:
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {probe}")
spark.sql(f"DROP SCHEMA IF EXISTS {probe}")
return True
except Exception as e:
if "Feature not supported on Apache Spark in Microsoft Fabric" in str(e):
return False
raise
if lakehouse_is_schema_enabled():
# Schema-enabled: create medallion schemas and write tables as `schema.table`.
spark.sql("CREATE SCHEMA IF NOT EXISTS bronze")
spark.sql("CREATE SCHEMA IF NOT EXISTS silver")
spark.sql("CREATE SCHEMA IF NOT EXISTS gold")
else:
# Not schema-enabled: skip CREATE SCHEMA (unsupported). Organize tables by name
# prefix instead (e.g. `bronze_orders` / `raw_orders`, `silver_orders` /
# `curated_orders`), or recreate the lakehouse with
# `creationPayload.enableSchemas: true` if schemas are required.
print("Lakehouse is not schema-enabled; skipping schema creation and using name prefixes.")
创建并刷新 Materialized Lake View（MLV）
-- See resources/materialized-lake-view-patterns.md for design guidance
-- and resources/mlv-incremental-refresh-patterns.md for refresh-readiness review.
-- Bronze/Silver/Gold schemas in a Lakehouse with schemas enabled
CREATE SCHEMA IF NOT EXISTS bronze;
CREATE SCHEMA IF NOT EXISTS silver;
CREATE SCHEMA IF NOT EXISTS gold;
-- A simple Silver MLV with data quality constraints
--
-- Prerequisite for incremental refresh: enable Change Data Feed (CDF) on every
-- source table the MLV reads from. Without CDF, optimal refresh can only choose
-- between no refresh (sources unchanged) and full refresh — never incremental.
-- See resources/mlv-incremental-refresh-patterns.md.
ALTER TABLE bronze.orders_raw SET TBLPROPERTIES (delta.enableChangeDataFeed = true);
CREATE OR REPLACE MATERIALIZED LAKE VIEW silver.orders_clean
(
CONSTRAINT valid_order_id CHECK (order_id IS NOT NULL) ON MISMATCH DROP
)
AS
SELECT
order_id,
customer_id,
CAST(order_ts AS TIMESTAMP) AS order_ts,
amount
FROM bronze.orders_raw;
-- Routine refresh is handled by the lakehouse Materialized lake views → Manage
-- schedule/lineage view; don't orchestrate from notebooks. The SQL form below is
-- documented only for forcing a one-time FULL recompute (troubleshooting / after
-- a correction). There is no documented SQL form for triggering incremental refresh.
REFRESH MATERIALIZED LAKE VIEW silver.orders_clean FULL;
创建 Lakehouse Livy 会话
# See SPARK-CONSUMPTION-CORE.md for Lakehouse Livy session configuration and management
# IMPORTANT: Body MUST be flat JSON with memory/cores — do NOT wrap in {"payload": ...}
cat > /tmp/body.json << 'EOF'
{"name": "dev-session", "driverMemory": "56g", "driverCores": 8, "executorMemory": "56g", "executorCores": 8, "conf": {"spark.dynamicAllocation.enabled": "true", "spark.fabric.pool.name": "Starter Pool"}}
EOF
az rest --method post --resource "https://api.fabric.microsoft.com" \
--url "https://api.fabric.microsoft.com/v1/workspaces/$workspace_id/lakehouses/$lakehouse_id/livyapi/versions/2023-12-01/sessions" \
--body @/tmp/body.json
Lakehouse Livy 会话正文 — 常见错误
- ❌
{"payload": {"kind": "pyspark"}}→ HTTP 500（包装格式错误，缺少必需字段）- ❌
{"kind": "pyspark"}→ HTTP 500（缺少driverMemory、executorMemory等字段）- ✅ 扁平 JSON，包含
name、driverMemory、driverCores、executorMemory、executorCores（使用 Starter Pool 时也可包含conf）
Spark 性能配置
有关特定工作负载的详细配置，请参阅 data-engineering-patterns.md 中的 Delta Lake 最佳实践。
快速参考：
# Write-heavy (Bronze): Disable V-Order, enable autoCompact
# Balanced (Silver): Enable V-Order, adaptive execution
# Read-heavy (Gold): Vectorized reads, optimal parallelism
# See data-engineering-patterns.md for complete config tables
重点：Spark/数据工程开发和笔记本代码编写所需的基本 CLI 模式，并通过智能路由指向专用资源。有关完整模式，请始终参考 COMMON-* 文件和资源文档。
评论
支持 Markdown