Eseguire la migrazione dal calcolo classico al calcolo serverless

Eseguire la migrazione dei carichi di lavoro dal calcolo classico al calcolo serverless. Il calcolo serverless gestisce automaticamente il provisioning, il ridimensionamento, gli aggiornamenti di runtime e l'ottimizzazione.

La maggior parte dei carichi di lavoro classici può eseguire la migrazione con modifiche minime o senza modifiche al codice. Questa pagina è incentrata su questi carichi di lavoro. Alcune funzionalità, ad esempio df.cache, non sono ancora supportate in serverless, ma non richiedono modifiche al codice una volta disponibili. Alcuni carichi di lavoro che dipendono da notebook R o Scala richiedono risorse di calcolo classiche e non potranno eseguire la migrazione a serverless. Per un elenco completo delle limitazioni correnti, vedere Limitazioni di calcolo serverless.

Passaggi per la migrazione

Per eseguire la migrazione dei carichi di lavoro dal calcolo classico al calcolo serverless, seguire questa procedura:

1. Verificare i prerequisiti: verificare che l'area di lavoro, la rete e l'accesso alle risorse di archiviazione cloud soddisfino i requisiti. Vedere Prima di iniziare.
2. Aggiornare il codice: apportare le modifiche necessarie al codice e alla configurazione. Vedere Aggiornare il codice.
3. Testare i carichi di lavoro: Convalidare la compatibilità e la correttezza prima della transizione. Vedere Testare i carichi di lavoro.
4. Scegliere una modalità di prestazioni: selezionare la modalità prestazioni più adatta ai requisiti del carico di lavoro. Vedere Scegliere una modalità di prestazioni.
5. Eseguire la migrazione in fasi: implementare in modo incrementale serverless, a partire da carichi di lavoro nuovi e a basso rischio. Vedere Eseguire la migrazione in fasi.
6. Monitorare i costi: Monitorare il consumo di DBU serverless e configurare avvisi. Vedi Monitorare i costi.

Prima di iniziare

Prima di iniziare la migrazione, potrebbe essere necessario aggiornare alcune configurazioni legacy nell'area di lavoro.

Verificare che l'area di lavoro sia in un'area supportata.

Aggiornare il codice

Le sezioni seguenti elencano le modifiche al codice e alla configurazione necessarie per rendere i carichi di lavoro compatibili con serverless.

L'accesso ai dati

I modelli di accesso ai dati legacy non sono supportati in serverless. Aggiornare il codice per usare invece Unity Catalog.

# 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 e codice

Alcune API e modelli di codice non sono supportate in serverless. Fare riferimento a questa tabella per verificare se è necessario aggiornare il codice.

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")

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")

Librerie e ambienti

È possibile gestire librerie e ambienti a livello di area di lavoro usando ambienti di base e a livello di notebook usando l'ambiente serverless del notebook.

Aggiungere pacchetti Python in requirements.txt per ambienti riproducibili. Specificare versioni dei pacchetti Python.

Trasmissione in diretta

I carichi di lavoro di streaming sono supportati in serverless, ma alcuni trigger non sono supportati. Aggiornare il codice per usare i trigger supportati.

Per lo streaming continuo, eseguire la migrazione alle pipeline dichiarative di Spark in modalità continua oppure utilizzare i job a pianificazione continua con AvailableNow. Per le origini di grandi dimensioni, impostare maxFilesPerTrigger o maxBytesPerTrigger per prevenire errori di esaurimento della memoria.

# 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))

Testare i carichi di lavoro

1. Test di compatibilità rapido: eseguire il carico di lavoro nel calcolo classico con la modalità di accesso Standard e Databricks Runtime 14.3 o versione successiva. Se l'esecuzione ha esito positivo, il carico di lavoro può eseguire la migrazione a serverless senza modifiche al codice.
2. Confronto A/B (consigliato per la produzione): eseguire lo stesso carico di lavoro in versione classica (controllo) e serverless (esperimento). Controllare le tabelle di output "diff" e verificarne la correttezza. Eseguire l'iterazione fino a quando gli output corrispondono.
3. Configurazioni temporanee: è possibile impostare temporaneamente configurazioni Spark supportate durante i test. Rimuoverli una volta che sono stabili.

Scegliere una modalità di prestazioni

I processi serverless e le pipeline supportano due modalità di prestazioni: standard e ottimizzati per le prestazioni. La modalità di prestazioni scelta dipende dai requisiti del carico di lavoro.

Eseguire la migrazione in fasi

1. Nuovi carichi di lavoro: avviare tutti i nuovi notebook e processi in serverless.
2. Carichi di lavoro a basso rischio: eseguire la migrazione di carichi di lavoro PySpark/SQL già in modalità di accesso standard e Databricks Runtime 14.3 o versione successiva.
3. Carichi di lavoro complessi: eseguire la migrazione dei carichi di lavoro che necessitano di modifiche al codice (riscrittura RDD, aggiornamenti DBFS, correzioni dei trigger).
4. Carichi di lavoro rimanenti: esaminare periodicamente man mano che le funzionalità si espandono.

Monitorare i costi

La fatturazione serverless si basa sul consumo DBU e non sul tempo di attività del cluster. Convalidare le aspettative sui costi con carichi di lavoro rappresentativi prima di eseguire la migrazione su larga scala. Per gli strumenti e le strategie per monitorare i costi serverless, vedere Monitorare il costo del calcolo serverless.

Risorse aggiuntive

• Procedure consigliate per il calcolo serverless: suggerimenti per l'ottimizzazione per carichi di lavoro serverless
• Limitazioni di calcolo serverless: elenco completo delle limitazioni correnti e delle funzionalità non supportate
• Configurare l'ambiente serverless: gestire librerie e dipendenze
• Configurazioni Spark supportate: Spark configurazioni disponibili nel serverless
• Spark Connect e spark classico: differenze comportamentali nell'architettura serverless
• Sicurezza di rete senza server: controller di rete, Private Link e configurazione del firewall
• Note sulla versione di calcolo serverless: Monitorare le nuove funzionalità quando vengono rilasciate
• Guida all'aggiornamento del catalogo Unity: Eseguire la migrazione da Hive Metastore a Unity Catalog

Per altre informazioni, vedere anche i post di blog seguenti:

• Che cos'è l'elaborazione serverless?: Panoramica delle funzionalità serverless e dei risultati dei clienti
• Evoluzione dell'ingegneria dei dati: come il calcolo serverless sta trasformando i notebook e i processi Lakeflow: come funzionano i processi Lakeflow e le pipeline senza server