從經典運算遷移到無伺服器運算

將你的工作負載從經典運算遷移到無伺服器運算。 無伺服器運算自動處理配置、擴展、執行時升級與優化。

大多數經典工作負載都能以極少甚至不需修改程式碼的方式遷移。 本頁聚焦於這些工作負載。 部分功能，如 df.cache，尚未支援無伺服器，但一旦啟用，將不再需要修改程式碼。 某些依賴 R 或 Scala 筆記本的工作負載需要經典運算，無法遷移到無伺服器。 關於目前的完整限制清單，請參見 無伺服器運算限制。

移轉步驟

要將您的工作負載從經典運算遷移到無伺服器運算，請依照以下步驟操作：

  1. 檢查先決條件：確認您的工作空間、網路及雲端儲存存取是否符合要求。 請參閱 開始之前。
  2. 更新程式碼：做必要的程式碼和設定變更。 請參見 更新你的程式碼。
  3. 測試你的工作負載：在切換前驗證相容性和正確性。 請參見 「測試你的工作負載」。
  4. 選擇效能模式：選擇最適合你工作負載需求的效能模式。 請參見 選擇表演模式。
  5. 分階段遷移：逐步推出無伺服器，從新且低風險的工作負載開始。 參見 分階段遷移。
  6. 監控成本：追蹤無伺服器 DBU 的使用情況並設定警示。 請參見 監控成本。

開始之前

在開始遷移之前，你可能需要更新工作空間中的一些舊有設定。

先決條件 Action 詳細資料
已為工作區啟用了 Unity Catalog 如有需要，請從 Hive Metastore 遷移 將Azure Databricks工作區升級為Unity Catalog
網路配置 以 NCC、Private Link 或防火牆規則取代 VPC 對等 無伺服器計算平面網路
雲端儲存存取 將舊有的資料存取模式替換為 Unity Catalog 的外部位置 使用 Unity 目錄連線到雲端物件儲存體

確認你的工作區是否位於 支援區域。

更新您的程式碼

以下章節列出為使工作負載與無伺服器相容所需的程式碼與設定變更。

數據存取

無伺服器系統不支援舊有的資料存取模式。 更新你的程式碼改用 Unity Catalog。

經典圖案 無伺服器架構替代 詳細資料
DBFS 路徑 （dbfs:/...） Unity 目錄卷 Unity Catalog 磁碟區是什麼？
Hive Metastore 資料表 Unity Catalog 表格（或稱 HMS Federation） 將Azure Databricks工作區升級為Unity Catalog
存儲帳戶憑證 Unity Catalog 外部位置 使用 Unity 目錄連線到雲端物件儲存體
客製化 JDBC JAR 檔案們 Lakehouse 聯盟 什麼是查詢同盟？

警告

DBFS 在無伺服器時存取有限。 遷移前請更新所有 dbfs:/ 通往 Unity 目錄卷的路徑。 欲了解更多資訊，請參閱 DBFS 中儲存的檔案遷移。

範例：替換 DBFS 路徑與 Hive Metastore 引用 
# Classic
df = spark.read.csv("dbfs:/mnt/datalake/data.csv", header=True)
df.write.parquet("dbfs:/mnt/output/results")
df = spark.table("my_database.my_table")

# Serverless
df = spark.read.csv("/Volumes/main/sales/raw_data/data.csv", header=True)
df.write.parquet("/Volumes/main/analytics/output/results")
df = spark.table("main.my_database.my_table")  # three-level namespace

API 與程式碼

某些 API 和程式碼模式在無伺服器上不被支援。 參考此表，看看你的程式碼是否需要更新。

經典圖案 無伺服器架構替代 詳細資料
RDD API （sc.parallelize， rdd.map） DataFrame API 比較 Spark Connect 與 Spark Classic
df.cache()、df.persist() 移除快取呼叫 無伺服器運算限制
spark.sparkContext、sqlContext 直接使用 spark （SparkSession） 比較 Spark Connect 與 Spark Classic
蜂巢變數 （${var}） SQL DECLARE VARIABLE 或 Python f-strings DECLARE VARIABLE
不支援的 Spark 設定 移除不支援的設定。 Serverless 大多數設定會自動調整。 為無伺服器筆記本與工作設定 Spark 屬性
範例：以資料框架取代 RDD 操作 
from pyspark.sql import functions as F

# sc.parallelize + rdd.map
# Classic:  rdd = sc.parallelize([1, 2, 3]); rdd.map(lambda x: x * 2).collect()
df = spark.createDataFrame([(1,), (2,), (3,)], ["value"])
result = df.select((F.col("value") * 2).alias("value")).collect()

# rdd.flatMap
# Classic:  sc.parallelize(["hello world"]).flatMap(lambda l: l.split(" ")).collect()
df = spark.createDataFrame([("hello world",)], ["line"])
words = df.select(F.explode(F.split("line", " ")).alias("word")).collect()

# rdd.groupByKey
# Classic:  rdd.groupByKey().mapValues(list).collect()
df = spark.createDataFrame([("a", 1), ("b", 2), ("a", 3)], ["key", "value"])
grouped = df.groupBy("key").agg(F.collect_list("value").alias("values")).collect()

# rdd.mapPartitions → applyInPandas
import pandas as pd
def process_group(pdf: pd.DataFrame) -> pd.DataFrame:
    return pd.DataFrame({"total": [pdf["id"].sum()]})
result = (spark.range(100).repartition(4)
    .groupBy(F.spark_partition_id())
    .applyInPandas(process_group, schema="total long").collect())

# sc.textFile → spark.read.text
df = spark.read.text("/Volumes/catalog/schema/volume/file.txt")
範例：替換 SparkContext 與快取 
from pyspark.sql.functions import broadcast

# sc.broadcast → broadcast join
result = main_df.join(broadcast(lookup_df), "key")

# sc.accumulator → DataFrame aggregation
total = df.agg(F.sum("amount")).collect()[0][0]

# sqlContext.sql → spark.sql
result = spark.sql("SELECT * FROM main.db.table")

# df.cache() → remove caching calls
# Materialize expensive intermediate results to Delta as a workaround:
df = spark.read.parquet(path)
result = df.filter("status = 'active'")
expensive_df.write.format("delta").mode("overwrite").saveAsTable("main.scratch.temp")
result = spark.table("main.scratch.temp")

函式庫與環境

你可以在工作區層級使用 基礎環境 管理函式庫和環境，並在筆記本層級使用筆記本的 無伺服器環境管理。

經典圖案 無伺服器架構替代 詳細資料
初始化腳本 無伺服器環境 設定無伺服器環境
叢集層級程式庫 限定於筆記本的或環境中的函式庫 設定無伺服器環境
Maven/JAR 函式庫 JAR工作支援;筆記本用的 PyPI 作業用的 JAR 任務
Docker 容器 無伺服器環境以滿足函式庫需求 設定無伺服器環境

在 requirements.txt 中鎖定 Python 封包，以便建立可重現的環境。 請參見 Specified Python 套件版本。

串流

串流工作負載在無伺服器架構上受支援，但不支援某些觸發事件。 更新你的程式碼，使用支援的觸發器。

火花觸發器 支援 Notes
Trigger.AvailableNow() 是的 Recommended
Trigger.Once() 是的 這已被取代。 請改用 Trigger.AvailableNow()。
Trigger.ProcessingTime(interval) No 傳回 INFINITE_STREAMING_TRIGGER_NOT_SUPPORTED。
Trigger.Continuous(interval) No 改用 Lakeflow 管線的連續模式
預設（非設定 .trigger()） No 省略 .trigger() 預設值為 ProcessingTime("0 seconds")，在無伺服器架構中不支援。 一定要明確設定 .trigger(availableNow=True) 。

若需連續串流，請在連續模式下遷移至 Spark 宣告式管線，或使用帶有 連續排程工作 的 AvailableNow。 對於大型來源，請設定 maxFilesPerTrigger 或 maxBytesPerTrigger 防止記憶體不足錯誤。

範例：修正串流觸發器 
# Classic (not supported on serverless — default trigger is ProcessingTime)
query = df.writeStream.format("delta").outputMode("append").start()

# Serverless (explicit AvailableNow trigger)
query = (df.writeStream.format("delta").outputMode("append")
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .start(output_path))
query.awaitTermination()

# With OOM prevention for large sources
query = (spark.readStream.format("delta")
    .option("maxFilesPerTrigger", 100)
    .option("maxBytesPerTrigger", "10g")
    .load(input_path)
    .writeStream.format("delta")
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .start(output_path))

測試你的工作負載

  1. 快速相容性測試：在經典運算上執行工作負載，並搭配 標準 存取模式及 Databricks Runtime 14.3 或以上版本。 如果執行成功，工作負載可以遷移到無伺服器，且不需修改程式碼。
  2. A/B 比較（建議用於生產環境）：在經典（控制）和無伺服器（實驗）上執行相同的工作負載。 比較輸出表格並驗證其正確性。 反覆迭代直到輸出匹配。
  3. 臨時設定：測試時可以暫時設定支援的 Spark 設定。 穩定後再取下。

選擇表演模式

無伺服器工作與管線支援兩種效能模式：標準與效能優化。 你選擇的效能模式取決於你的工作負載需求。

模式 可用性 Startup 最適合用於
標準 工作與 Lakeflow 管線 4-6分鐘 成本敏感批次
效能優化 筆記本、作業、Lakeflow 管線 秒 互動式，對延遲敏感

分階段遷徙

  1. 新增工作負載：將所有新筆記本和工作都用無伺服器模式啟動。
  2. 低風險工作負載：遷移已在標準存取模式及 Databricks 執行環境 14.3 以上的 PySpark/SQL 工作負載。
  3. 複雜工作負載：遷移需要修改程式碼的工作負載（如 RDD 重寫、DBFS 更新、觸發修正）。
  4. 剩餘工作量：在能力擴展的同時定期檢視。

成本監控

無伺服器計費是以 DBU 消耗為基礎，而非叢集運作時間。 在大規模遷移前，請以具代表性的工作負載驗證成本預期。 關於監控無伺服器成本的工具與策略，請參見 「監控無伺服器運算成本」。

其他資源

您也可以參考以下部落格文章以獲得更多資訊：

  • Last updated on