Spark ConnectとSpark Classicを比較する

Spark Connect は、Apache Spark 内の gRPC ベースのプロトコルであり、クライアントアプリケーションがリモート Spark サーバーと通信する方法を指定します。DataFrame API を使用して Spark ワークロードをリモートで実行できます。

Spark Connect は次のような場合に使用されます。

標準コンピュート上のDatabricks Runtimeバージョン 13.3 以降を使用したScalaノートブック
標準コンピュート上のDatabricks Runtimeバージョン 14.3 以降を使用したPythonノートブック
サーバーレスコンピュート
Databricks Connect
環境バージョンが構成されたLakeFlow Pipelines

Spark Connect と Spark Classic はどちらも変換に遅延実行を利用しますが、既存のコードを Spark Classic から Spark Connect に移行するとき、または両方で動作する必要があるコードを作成するときに、予期しない動作やパフォーマンスの問題を回避するために知っておくべき重要な違いがあります。

怠惰 vs 熱心

Spark Connect と Spark Classic の主な違いは、次の表にまとめられているように、Spark Connect では分析と名前解決が実行時に延期されることです。

観点	Sparkクラシック	Sparkコネクト
クエリ実行	怠け者	怠け者
スキーマ分析	熱心な	怠け者
スキーマアクセス	ローカル	RPC をトリガーし、最初のアクセス時にスキーマをキャッシュします。
一時的なビュー	計画が組み込まれている	名前検索
UDFシリアル化	創造時	実行時

観点	Sparkクラシック	Sparkコネクト
クエリ実行	怠け者	怠け者
スキーマ分析	熱心な	怠け者
スキーマアクセス	ローカル	RPC をトリガーし、最初のアクセス時にスキーマをキャッシュします。
一時的なビュー	計画が組み込まれている	名前検索
UDFシリアル化	創造時	実行時

クエリ実行

Spark Classic と Spark Connect はどちらも、クエリ実行に同じ遅延実行モデルに従います。

Spark Classic では、DataFrame 変換 ( filterやlimitなど) は遅延されます。つまり、すぐに実行されるのではなく、論理プランにエンコードされます。実際の計算はアクション ( show() 、 collect()など) によってのみトリガーされます。

Spark Connect も同様の遅延評価モデルに従います。変換はクライアント側で構築され、未解決のプランとしてサーバーに送信されます。その後、アクションが呼び出されると、サーバーは必要な分析と実行を実行します。

観点	Sparkクラシック	Sparkコネクト
変換: `df.filter(...)` 、 `df.select(...)` 、 `df.limit(...)`	遅延実行	遅延実行
SQL クエリ: `spark.sql("select …")`	遅延実行	遅延実行
アクション: `df.collect()` 、 `df.show()`	熱心な実行	熱心な実行
SQLコマンド: `spark.sql("insert …")` 、 `spark.sql("create …")`	熱心な実行	熱心な実行

観点	Sparkクラシック	Sparkコネクト
変換: `df.filter(...)` 、 `df.select(...)` 、 `df.limit(...)`	遅延実行	遅延実行
SQL クエリ: `spark.sql("select …")`	遅延実行	遅延実行
アクション: `df.collect()` 、 `df.show()`	熱心な実行	熱心な実行
SQLコマンド: `spark.sql("insert …")` 、 `spark.sql("create …")`	熱心な実行	熱心な実行

スキーマ分析

Spark Classic は、論理プランの構築中に積極的に分析を実行します。この分析フェーズでは、未解決のプランを完全に解決された論理プランに変換し、操作が Spark によって実行できることを確認します。この作業を熱心に行うことで得られる主な利点の 1 つは、間違いがあったときにユーザーが即座にフィードバックを受け取れることです。たとえば、 spark.sql("select 1 as a, 2 as b").filter("c > 1")を実行すると、列cが見つからないことを示すエラーがすぐにスローされます。

Spark Connect は、クライアントが変換中に未解決のプランを構築し、その分析を延期する点で Classic と異なります。スキーマへのアクセス、プランの説明、DataFrame の永続化、アクションの実行など、解決済みのプランを必要とする操作を行うと、クライアントは RPC 経由で未解決のプランをサーバーに送信します。次に、サーバーは完全な分析を実行して、解決された論理プランを取得し、操作を実行します。たとえば、 spark.sql("select 1 as a, 2 as b").filter("c > 1")未解決のプランがクライアント側のみにあるためエラーはスローされませんが、 df.columnsまたはdf.show()では未解決のプランが分析のためにサーバーに送信されるためエラーがスローされます。

クエリ実行とは異なり、Spark Classic と Spark Connect ではスキーマ分析がいつ実行されるかが異なります。

観点	Sparkクラシック	Sparkコネクト
変換: `df.filter(...)` 、 `df.select(...)` 、 `df.limit(...)`	熱心な	怠け者
スキーマアクセス: `df.columns` 、 `df.schema` 、 `df.isStreaming`	熱心な	熱心な Spark Classicとは異なり、分析RPCリクエストをトリガーします。
アクション: `df.collect()` 、 `df.show()`	熱心な	熱心な
DataFrames の依存セッション状態: UDF、一時ビュー、構成	熱心な	怠け者 DataFrameのプラン実行中に評価される
一時ビューの依存セッション状態: UDF、他の一時ビュー、構成	熱心な	熱心な一時ビューを作成すると分析が即時に開始されます

観点	Sparkクラシック	Sparkコネクト
変換: `df.filter(...)` 、 `df.select(...)` 、 `df.limit(...)`	熱心な	怠け者
スキーマアクセス: `df.columns` 、 `df.schema` 、 `df.isStreaming`	熱心な	熱心な Spark Classicとは異なり、分析RPCリクエストをトリガーします。
アクション: `df.collect()` 、 `df.show()`	熱心な	熱心な
DataFrames の依存セッション状態: UDF、一時ビュー、構成	熱心な	怠け者 DataFrameのプラン実行中に評価される
一時ビューの依存セッション状態: UDF、他の一時ビュー、構成	熱心な	熱心な一時ビューを作成すると分析が即時に開始されます

ベストプラクティス

遅延分析と積極的分析の違いは、予期しない動作やパフォーマンスの問題、具体的には一時ビュー名の上書き、UDF での外部変数のキャプチャ、エラー検出の遅延、新しいDataFramesでの過度のスキーマアクセスなどによって発生する問題を回避するために従うべきベストプラクティスがいくつかあることを意味します。

一意の一時ビュー名を作成する

Spark Connect では、DataFrame には名前による一時ビューへの参照のみが保存されます。その結果、後で一時ビューが置き換えられると、実行時に名前でビューが検索されるため、DataFrame 内のデータも変更されます。

この動作は、作成時に一時ビューの論理プランがデータフレームのプランに埋め込まれる Spark Classic とは異なります。その後の一時ビューの置き換えは、元のデータフレームに影響しません。

違いを軽減するには、常に一意の一時ビュー名を作成します。たとえば、ビュー名に UUID を含めます。これにより、以前に登録された一時ビューを参照する既存のDataFramesへの影響が回避されます。

Python
Scala

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)

UDF定義をラップする

UDF が変更可能な外部変数に依存することは、暗黙的な依存関係を導入し、非決定論的な動作につながり、構成可能性を低下させる可能性があるため、一般的に悪い習慣と考えられています。ただし、そのようなパターンがある場合は、次の点に注意してください。

Spark Connect では、Python UDF は遅延します。シリアル化と登録は実行時まで延期されます。次の例では、 show()が呼び出されたときにのみ、UDF がシリアル化され、実行のために Spark クラスターにアップロードされます。

Python
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

この動作は、UDF が積極的に作成される Spark Classic とは異なります。Spark Classic では、UDF 作成時のxの値がキャプチャされるため、その後のxの変更は既に作成されている UDF には影響しません。

UDF が依存する外部変数の値を変更する必要がある場合は、関数ファクトリ (早期バインディングによるクロージャ) を使用して変数値を正しくキャプチャします。具体的には、UDF の作成をヘルパー関数でラップして、従属変数の値を取得します。

Python
Scala

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

UDF 定義を別の関数 ( make_udf ) 内にラップすることで、 xの現在の値が引数として渡される新しいスコープを作成します。これにより、生成された各 UDF に、UDF の作成時にバインドされたフィールドの独自のコピーが含まれるようになります。

エラー検出のためのトリガーイーガー分析

次のエラー処理は、例外をすぐにスローできるようにする積極的な分析を実行するため、Spark Classic で役立ちます。ただし、Spark Connect では、このコードは分析をトリガーせずにローカルの未解決プランを構築するだけなので、問題は発生しません。

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

コードが分析例外に依存しており、それをキャッチしたい場合は、たとえばdf.columns 、 df.schema 、またはdf.collect()を使用して、積極的な分析をトリガーできます。

Python
Scala

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

過剰な分析要求を避ける

多数のDataFramesに対する分析リクエストを回避すると、パフォーマンスが向上します。

新しいDataFramesステップごとに作成し、反復ごとにそのスキーマにアクセスする

大量の新しいDataFramesを作成する場合、それらに対して積極的な分析をトリガーする呼び出し ( df.columns 、 df.schemaなど) を過度に使用しないようにしてください。同じDataFrameのスキーマに複数回アクセスできますが、新しく作成された多数のDataFramesで分析をトリガーすると、パフォーマンスに影響します。

たとえば、ループ内で DataFrame に列を反復的に追加し、各列を追加する前にその列が既に存在するかどうかを確認する場合、新しく作成された各 DataFrame でdf.columnsを呼び出すと、反復ごとに分析リクエストがトリガーされます。これを回避するには、DataFrame のスキーマに繰り返しアクセスするのではなく、列名を追跡するためのセットを維持します。

Python
Scala

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

多数の中間DataFramesスキーマにアクセスしないようにする

もう一つの同様のケースは、不要な中間DataFrames大量に作成して分析することです。次の場合、構造体型の各列からフィールド名を抽出するには、中間DataFrames作成する代わりに、 DataFrameのスキーマから直接StructTypeフィールド情報を取得します。

Python
Scala

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)

怠惰 vs 熱心​

クエリ実行​

スキーマ分析​

ベストプラクティス​

一意の一時ビュー名を作成する​

UDF定義をラップする​

エラー検出のためのトリガーイーガー分析​

過剰な分析要求を避ける​

新しいDataFramesステップごとに作成し、反復ごとにそのスキーマにアクセスする​

多数の中間DataFramesスキーマにアクセスしないようにする​