Spark Connect: gRPC thin client и удалённое подключение без локального Spark

// Упрощённый пример из spark/connect/proto/relations.proto
message Relation {
  RelationCommon common = 1;
  oneof rel_type {
    Read read = 2;
    Project project = 3;
    Filter filter = 4;
    Join join = 5;
    Aggregate aggregate = 6;
    Sort sort = 7;
    // ...
  }
}

message Filter {
  Relation input = 1;   // входное отношение (другая Relation)
  Expression condition = 2;  // условие фильтрации
}

# Запустить Spark Connect Server (Spark 3.4+)
./sbin/start-connect-server.sh \
  --packages org.apache.spark:spark-connect_2.12:3.5.0 \
  --master yarn \
  --conf spark.executor.instances=10

# Или через spark-submit:
spark-submit \
  --packages org.apache.spark:spark-connect_2.12:3.5.0 \
  --class org.apache.spark.sql.connect.service.SparkConnectServer \
  # (класс запускается автоматически через скрипт)

# Классический PySpark (требует JVM + полный Spark):
spark = SparkSession.builder \
    .master("yarn") \
    .appName("MyJob") \
    .getOrCreate()

# Spark Connect (только Python-пакет, без JVM):
spark = SparkSession.builder \
    .remote("sc://spark-server.internal:15002") \
    .getOrCreate()

# Строка подключения:
# sc://host:port[/;param=value;...]
# sc://spark-server:15002
# sc://spark-server:15002/;token=my-auth-token
# sc://spark-server:15002/;use_ssl=true

# Минимальная установка для Spark Connect (без полного Spark):
pip install pyspark==3.5.0
# Или если нужен только клиент (Databricks):
pip install databricks-connect==14.0

# В классическом PySpark:
spark.sparkContext.addPyFile("mymodule.py")
spark.sparkContext.addFile("config.json")

# В Spark Connect:
spark.addArtifacts("mymodule.py")        # Python-модуль
spark.addArtifacts("config.json")        # произвольный файл
spark.addArtifacts("mypackage.whl")      # Python-пакет
# Файлы передаются на сервер через gRPC, а затем раздаются Executor'ам

# Подключение с TLS и токеном аутентификации:
spark = SparkSession.builder \
    .remote("sc://spark-cluster:15002/;use_ssl=true;token=Bearer_TOKEN_HERE") \
    .getOrCreate()

# Раньше: нужен локальный Spark + JVM + JAVA_HOME
# Сейчас: только pip install pyspark

spark = SparkSession.builder \
    .remote("sc://dev-spark-cluster:15002") \
    .getOrCreate()

df = spark.read.parquet("s3://data-lake/events/")
df.filter(col("date") >= "2024-01-01").show()

# Dockerfile: FROM python:3.11-slim
# pip install pyspark fastapi uvicorn
# Никакого JVM!

from fastapi import FastAPI
from pyspark.sql import SparkSession

app = FastAPI()

@app.on_event("startup")
async def startup():
    global spark
    spark = SparkSession.builder \
        .remote("sc://spark-cluster:15002") \
        .getOrCreate()

@app.get("/analytics/{user_id}")
async def get_user_stats(user_id: str):
    result = spark.sql(f"""
        SELECT region, count(*) as events
        FROM events
        WHERE user_id = '{user_id}'
        GROUP BY region
    """).collect()
    return {"stats": [row.asDict() for row in result]}

# JupyterHub без Spark-инсталляции на каждом pod'е
# Единый Spark Connect Server на кластере
# Все ноутбуки подключаются к нему
spark = SparkSession.builder \
    .remote("sc://shared-spark:15002/;token=USER_TOKEN") \
    .getOrCreate()

# Изоляция: каждый пользователь видит свои temp views
spark.sql("CREATE TEMP VIEW my_data AS SELECT ...")

# Сервер: Spark 3.5 на Python 3.9
# Клиент: pyspark 3.5, Python 3.11 - совместимо!

# Правило совместимости:
# - Минорные версии: 3.5.x клиент ↔ 3.5.y сервер - любая комбинация
# - Мажорные версии: 3.x ↔ 4.x - обратная совместимость не гарантирована

# Логи Spark Connect Server (внутри Driver лог-файлов):
grep "SparkConnectService\|ConnectSession" spark-driver.log

# Типичные записи:
# INFO SparkConnectService: Starting SparkConnect server on port 15002
# INFO ConnectSessionManager: New session: user=alice, sessionId=abc123
# INFO ConnectSessionManager: Session abc123 closed after 3600s

import logging

# Включить gRPC-логирование
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("grpc").setLevel(logging.DEBUG)

spark = SparkSession.builder.remote("sc://server:15002").getOrCreate()

# Проверить соединение
print(spark.version)             # версия сервера
print(spark.sparkContext.master) # "spark://server:15002" (Connect URL)

# Требования: Java 11+, Spark 3.5+ distribution
export SPARK_HOME=/opt/spark-3.5.0

# Запуск в local mode
$SPARK_HOME/sbin/start-connect-server.sh \
  --master local[4] \
  --packages org.apache.spark:spark-connect_2.12:3.5.0

# Проверить что слушает порт:
ss -tlnp | grep 15002

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, avg

# Подключение через Spark Connect
spark = SparkSession.builder \
    .remote("sc://localhost:15002") \
    .getOrCreate()

print(f"Подключён к Spark {spark.version}")

# DataFrame API работает как обычно
df = spark.range(1_000_000) \
    .selectExpr("id", "id % 100 AS bucket", "rand() * 100 AS value")

result = df.groupBy("bucket") \
    .agg(count("*").alias("cnt"), avg("value").alias("avg_value")) \
    .orderBy("bucket")

result.show(5)
# Результат приходит через gRPC как Arrow batches

from pyspark.sql.functions import udf, pandas_udf
from pyspark.sql.types import StringType
import pandas as pd

# Python UDF - работает через CloudPickle → gRPC → Executor
@udf(StringType())
def classify(v):
    if v > 75:
        return "high"
    elif v > 25:
        return "medium"
    return "low"

# Pandas UDF - работает через Arrow → gRPC → Executor
@pandas_udf(StringType())
def classify_batch(s: pd.Series) -> pd.Series:
    return s.map(lambda v: "high" if v > 75 else "medium" if v > 25 else "low")

df.withColumn("class", classify("value")).show(3)
df.withColumn("class", classify_batch("value")).show(3)

import time

# Spark Connect: цепочка transformations - локально, Action - один gRPC
spark_connect = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()

t0 = time.time()
for _ in range(100):
    # Каждый вызов - локальное построение дерева (без сети)
    _ = spark_connect.range(100).filter(col("id") > 50)
# Ни одного сетевого вызова - всё локально!
print(f"Spark Connect (100 transformations, no action): {(time.time()-t0)*1000:.1f}ms")

# Один Action - один gRPC
t0 = time.time()
spark_connect.range(100).filter(col("id") > 50).count()
print(f"Spark Connect (1 action): {(time.time()-t0)*1000:.1f}ms")

# В классическом Py4J каждый .filter(), .select() - это TCP round-trip