Porovnání Spark Connect s klasickým Sparkem

Spark Connect je protokol založený na gRPC v Rámci Apache Sparku, který určuje, jak klientská aplikace může komunikovat se vzdáleným Spark Serverem. Umožňuje vzdálené spouštění úloh Sparku pomocí rozhraní DATAFrame API.

Spark Connect se používá v následujících:

  • Notebooky Scala s Databricks Runtime verze 13.3 a vyšší na standardním výpočetním prostředí.
  • Notebooky Pythonu s Databricks Runtime verze 14.3 a vyšší na standardním výpočetním prostředí.
  • Výpočetní prostředí bez serveru
  • Databricks Connect
  • Kanály Lakeflow s nakonfigurovanými verzemi prostředí

I když Spark Connect i Spark Classic využívají opožděné spouštění transformací, existují důležité rozdíly, které je potřeba vědět, aby nedocházelo k neočekávaným chování a problémům s výkonem při migraci existujícího kódu ze Spark Classic na Spark Connect nebo při psaní kódu, který musí fungovat s oběma.

Opožděný vs. dychtivý

Hlavní rozdíl mezi Spark Connect a Spark Classic spočívá v tom, že Spark Connect odkládá analýzu a rozpoznání názvů na čas provedení, jak je shrnuto v následující tabulce.

Aspekt Spark Classic Spark Connect
Provedení dotazu Líný Líný
Analýza schématu Dychtivý Líný
Přístup ke schématu Local Aktivuje RPC a ukládá schéma do mezipaměti při prvním přístupu.
Dočasná zobrazení Vložený plán Vyhledávání názvů
Serializace UDF Při vytváření Při spuštění

Provedení dotazu

Spark Classic i Spark Connect se řídí stejným opožděným prováděcím modelem pro provádění dotazů.

V klasickém Sparku jsou transformace datového rámce (například filter a limit) opožděné. To znamená, že se nespouštějí okamžitě, ale jsou kódovány v logickém plánu. Skutečný výpočet se aktivuje pouze pomocí akce (například show(), collect()).

Spark Connect se řídí podobným zpožděným vyhodnocovacím modelem. Transformace se vytvářejí na straně klienta a odesílají se jako nevyřešené plány na server. Server pak provede potřebnou analýzu a spuštění při volání akce.

Aspekt Spark Classic Spark Connect
Transformace: df.filter(...), df.select(...), df.limit(...) Opožděné spuštění Opožděné spuštění
Dotazy SQL: spark.sql("select …") Opožděné spuštění Opožděné spuštění
Akce: df.collect(), df.show() Okamžité spuštění Okamžité spuštění
Příkazy SQL: spark.sql("insert …"), spark.sql("create …") Okamžité spuštění Okamžité spuštění

Analýza schématu

Spark Classic provádí analýzu bez prodlevy během sestavování logického plánu. Tato fáze analýzy převede nevyřešený plán na plně vyřešený logický plán a ověří, že operaci může spustit Spark. Jednou z klíčových výhod dychtivého provádění této práce je, že uživatelé dostanou okamžitou zpětnou vazbu, když dojde k chybě. Například spuštění spark.sql("select 1 as a, 2 as b").filter("c > 1") okamžitě vyvolá chybu, která naznačuje, že sloupec c nelze najít.

Spark Connect se liší od modelu Classic, protože klient během procesu transformace vytváří nevyřešené plány a odkládá jejich analýzu. Jakákoli operace, která vyžaduje vyřešený plán, například přístup ke schématu, vysvětlení plánu, zachování datového rámce nebo spuštění akce, způsobí, že klient odešle nevyřešené plány na server přes RPC. Server pak provede úplnou analýzu, aby získal vyřešený logický plán a provedl operaci. Například nebude vyvolána žádná chyba, protože spark.sql("select 1 as a, 2 as b").filter("c > 1") nevyřešený plán je pouze klientsky, ale na df.columns nebo df.show() dojde k chybě, protože nevyřešený plán se odešle na server k analýze.

Na rozdíl od spouštění dotazů se Spark Classic a Spark Connect liší, když dojde k analýze schématu.

Aspekt Spark Classic Spark Connect
Transformace: df.filter(...), df.select(...), df.limit(...) Dychtivý Líný
Přístup ke schématu: df.columns, df.schemadf.isStreaming Dychtivý Dychtivý
Aktivuje požadavek RPC analýzy, na rozdíl od Klasického Sparku.
Akce: df.collect(), df.show() Dychtivý Dychtivý
Závislý stav relace datových rámců: UDF, dočasná zobrazení, konfigurace Dychtivý Líný
Vyhodnoceno během provádění plánu datového rámce
Závislý stav relace dočasných zobrazení: Uživatelsky Definované Funkce (UDF), jiná dočasná zobrazení, konfigurace Dychtivý Dychtivý
Analýza se aktivuje okamžitě při vytváření dočasného zobrazení.

Osvědčené postupy

Rozdíl mezi opožděnou a dychtivou analýzou znamená, že existuje několik osvědčených postupů, které je potřeba dodržovat, aby nedocházelo k neočekávaným chování a problémům s výkonem, konkrétně problémům způsobeným přepsáním názvů dočasných zobrazení, zachycením externích proměnných v UDF, zpožděným zjišťováním chyb a nadměrným přístupem ke schématu v nových datových rámcích.

Vytvoření jedinečných dočasných názvů zobrazení

V nástroji Spark Connect datový rámec ukládá pouze odkaz na dočasné zobrazení podle názvu. V důsledku toho, pokud je dočasné zobrazení později nahrazeno, data v datovém rámci se také změní, protože hledá zobrazení podle názvu v době spuštění.

Toto chování se liší od modelu Spark Classic, kde je logický plán dočasného zobrazení vložen do plánu datového rámce při vytváření. Jakákoli následná náhrada dočasného zobrazení nemá vliv na původní datový rámec.

Pokud chcete tento rozdíl zmírnit, vždy vytvořte jedinečné dočasné názvy zobrazení. Do názvu zobrazení zadejte například UUID. Tím se zabrání ovlivnění existujících datových rámců, které odkazují na dříve zaregistrované dočasné zobrazení.

Python

import uuid
def create_temp_view_and_create_dataframe(x):
  temp_view_name = f"`temp_view_{uuid.uuid4()}`"  # Use a random name to avoid conflicts.
  spark.range(x).createOrReplaceTempView(temp_view_name)
  return spark.table(temp_view_name)

df10 = create_temp_view_and_create_dataframe(10)
assert len(df10.collect()) == 10

df100 = create_temp_view_and_create_dataframe(100)
assert len(df10.collect()) == 10  # It works as expected now.
assert len(df100.collect()) == 100

Scala

import java.util.UUID

def createTempViewAndDataFrame(x: Int) = {
  val tempViewName = s"`temp_view_${UUID.randomUUID()}`"
  spark.range(x).createOrReplaceTempView(tempViewName)
  spark.table(tempViewName)
}

val df10 = createTempViewAndDataFrame(10)
assert(df10.collect().length == 10)

val df100 = createTempViewAndDataFrame(100)
assert(df10.collect().length == 10) // Works as expected
assert(df100.collect().length == 100)

Zabalení definic uživatelem definovaných funkcí (UDF)

Obecně se považuje za špatný postup, aby uživatelsky definované funkce závisely na proměnlivých externích proměnných, protože to zavádí implicitní závislosti, může vést k nedeterministickému chování a snižuje složitost. Pokud však takový vzor máte, mějte na paměti následující gotcha:

V Nástroji Spark Connect jsou uživatelské funkce definované v Pythonu zpracovávány líně. Jejich serializace a registrace jsou odloženy do doby provádění. V následujícím příkladu je uživatelsky definovaná funkce serializována a nahrána do clusteru Spark pro spuštění, když je volána show().

from pyspark.sql.functions import udf

x = 123

@udf("INT")
def foo():
  return x


df = spark.range(1).select(foo())
x = 456
df.show() # Prints 456

Toto chování se liší od klasického Sparku, kde se dychtivě vytvářejí funkce definované uživatelem. V klasickém Sparku je hodnota x zaznamenána při vytvoření UDF, takže následné změny v x nemají vliv na již vytvořenou uživatelsky definovanou funkci.

Pokud potřebujete upravit hodnotu externích proměnných, na které závisí UDF, použijte uzávěr s časnou vazbou ke správnému zachycení hodnot proměnných. Konkrétně zabalte vytváření uživatelem definované funkce do pomocné funkce pro zachycení hodnoty závislé proměnné.

Python

from pyspark.sql.functions import udf

def make_udf(value):
  def foo():
    return value
  return udf(foo)


x = 123
foo_udf = make_udf(x)
x = 456
df = spark.range(1).select(foo_udf())
df.show() # Prints 123 as expected

Scala

def makeUDF(value: Int) = udf(() => value)

var x = 123
val fooUDF = makeUDF(x)  // Captures the current value
x = 456
val df = spark.range(1).select(fooUDF())
df.show() // Prints 123 as expected

Zabalením definice UDF do jiné funkce (make_udf) vytvoříme nový obor, ve kterém se aktuální hodnota x předá jako argument. Tím se zajistí, že každá vygenerovaná funkce definovaná uživatelem má svou vlastní kopii pole svázaného v okamžiku vytvoření definovaného uživatelem.

Aktivace okamžité analýzy pro detekci chyb

Následující zpracování chyb je užitečné ve Sparku Classic, protože provádí dychtivou analýzu, která umožňuje rychlé vyvolání výjimek. V nástroji Spark Connect ale tento kód nezpůsobuje žádný problém, protože vytváří pouze místní nevyřešený plán bez aktivace analýzy.

df = spark.createDataFrame([("Alice", 25), ("Bob", 30)], ["name", "age"])

try:
  df = df.select("name", "age")
  df = df.withColumn(
      "age_group",
      when(col("age") < 18, "minor").otherwise("adult"))
  df = df.filter(col("age_with_typo") > 6) # The use of non-existing column name will not throw analysis exception in Spark Connect
except Exception as e:
  print(f"Error: {repr(e)}")

Pokud váš kód spoléhá na výjimku analýzy a chcete ji zachytit, můžete aktivovat dychtivou analýzu, například pomocí df.columns, df.schemanebo df.collect().

Python

try:
  df = ...
  df.columns # This will trigger eager analysis
except Exception as e:
  print(f"Error: {repr(e)}")

Scala

import org.apache.spark.SparkThrowable
import org.apache.spark.sql.functions._

val df = spark.createDataFrame(Seq(("Alice", 25), ("Bob", 30))).toDF("name", "age")

try {
  val df2 = df.select("name", "age")
    .withColumn("age_group", when(col("age") < 18, "minor").otherwise("adult"))
    .filter(col("age_with_typo") > 6)
  df2.columns // Trigger eager analysis to catch the error
} catch {
  case e: SparkThrowable => println(s"Error: ${e.getMessage}")
}

Vyhněte se příliš mnoha žádostem o analýzu.

Pokud se vyhnete analytickým požadavkům na velký počet datových rámců, můžete zlepšit výkon.

Podrobné vytvoření nových datových rámců a přístup k jejich schématu v každé iteraci

Když vytváříte velký počet nových datových rámců, vyhněte se nadměrnému využití volání, která na nich aktivují dychtivou analýzu (například df.columns). df.schema Ke schématu stejného datového rámce můžete přistupovat vícekrát, ale aktivace analýzy mnoha nově vytvořených datových rámců ovlivní výkon.

Například při iterativním přidávání sloupců do datového rámce uvnitř smyčky a kontrole, zda již každý sloupec existuje před jeho přidáním, vyvolá použití df.columns na každý nově vytvořený datový rámec při každé iteraci požadavek na analýzu. Abyste tomu předešli, udržujte sadu pro sledování názvů sloupců místo opakovaného přístupu ke schématu datového rámce.

Python
df = spark.range(10)
columns = set(df.columns) # Maintain the set of column names
for i in range(200):
  new_column_name = str(i)
  # if new_column_name not in df.columns:  # Bad practice. The `df.columns` call causes an analysis request on the newly created DataFrame in every iteration.
  if new_column_name not in columns:  # Check the set without triggering analysis
    df = df.withColumn(new_column_name, F.col("id") + i)
    columns.add(new_column_name)
df.show()
Scala
import org.apache.spark.sql.functions._

var df = spark.range(10).toDF
val columns = scala.collection.mutable.Set(df.columns: _*)
for (i <- 0 until 200) {
  val newColumnName = i.toString
  // if (!df.columns.contains(newColumnName)) {  // Bad practice. The `df.columns` call causes an analysis request on the newly created DataFrame in every iteration.
  if (!columns.contains(newColumnName)) {  // Check the set without triggering analysis
    df = df.withColumn(newColumnName, col("id") + i)
    columns.add(newColumnName)
  }
}
df.show()

Vyhněte se přístupu ke schématům pro velký počet mezilehlých datových rámců.

Dalším podobným případem je vytvoření velkého počtu nepotřebných mezilehlých datových rámců a jejich analýza. Pokud chcete extrahovat názvy polí z každého sloupce typu struktury, získejte StructType informace o polích přímo ze schématu datového rámce místo vytváření zprostředkujících datových rámců.

Python
from pyspark.sql.types import StructType

df = ...
struct_column_fields = {
    # column_schema.name: df.select(column_schema.name + ".*").columns  # Bad practice. This creates an intermediate DataFrame and triggers an analysis request for each StructType column.
    column_schema.name: [f.name for f in column_schema.dataType.fields]  # Access StructType fields directly from the schema, avoiding analysis on intermediate DataFrames.
    for column_schema in df.schema
    if isinstance(column_schema.dataType, StructType)
}
print(struct_column_fields)
Scala
import org.apache.spark.sql.types.StructType

df = ...
val structColumnFields = df.schema.fields
  .filter(_.dataType.isInstanceOf[StructType])
  .map { field =>
    // field.name -> df.select(field.name + ".*").columns  // Bad practice. This creates an intermediate DataFrame and triggers analysis for each StructType column.
    field.name -> field.dataType.asInstanceOf[StructType].fields.map(_.name)  // Access StructType fields directly from the schema, avoiding analysis on intermediate DataFrames.
  }
  .toMap
println(structColumnFields)
  • Last updated on