connect

Spark Connect is a gRPC layer that lets thin clients drive a remote Spark driver without embedding the JVM. Introduced in Spark 3.4 and made GA in 3.5, it now powers PySpark in Connect mode, the JVM Connect client, and a growing list of language clients (Go, Swift, Rust have community implementations on top of the same protocol).

Purpose

• Decouple the client from the driver so that SDK upgrades do not require restarting a Spark cluster.
• Allow lightweight clients (a notebook, an IDE, a CLI) to talk to a long-running Spark service.
• Provide a single wire format (protobuf) that captures the DataFrame API, SQL, ML pipelines, and declarative pipelines.

Directory layout

sql/connect/
  bin/                     - start-/stop-connect-server scripts
  client/                  - JVM Connect client (Scala). Mirrors the classic DataFrame API.
  common/                  - protobuf definitions and the shared client-server types
    src/main/protobuf/spark/connect/
      base.proto           - SparkSession-level service (Plan, ExecutePlan, ...)
      relations.proto      - the LogicalPlan-equivalent message tree
      expressions.proto    - the Expression-equivalent message tree
      commands.proto       - SQL/DDL commands
      catalog.proto        - catalog operations (databases, tables, functions)
      ml.proto, ml_common.proto - MLlib operations
      pipelines.proto      - declarative pipelines
      common.proto         - shared atoms (DataType, Origin, ...)
      example_plugins.proto - plugin extension example
  docs/                    - protocol notes
  server/                  - the gRPC server that runs in the driver JVM
  shims/                   - per-Spark-version compatibility shims for the client

The top-level Python client lives in python/pyspark/sql/connect/, python/pyspark/ml/connect/, etc.

Key abstractions

Where	What it is
SparkConnectService (sql/connect/server/.../SparkConnectService.scala)	The gRPC service entry point.
SparkConnectPlanner (sql/connect/server/.../planner/SparkConnectPlanner.scala)	Translates protobuf relations into Catalyst plans.
SparkConnectAnalyzer / SparkConnectExecutor	Run analyzed plans in an embedded SparkSession.
SparkSession (Scala connect client) (sql/connect/client/.../SparkSession.scala)	Mirror of the classic API; sends requests over gRPC.
pyspark.sql.connect.session.SparkSession (python/pyspark/sql/connect/session.py)	Python connect entry point.
Plan, Relation, Expression	The wire-format messages defined in protobuf.
ArtifactManager (sql/core/.../artifact/)	Tracks per-session jars/files transmitted by the client.

Request / response flow

sequenceDiagram
    participant C as Client (Python / JVM)
    participant S as Connect server (gRPC)
    participant P as SparkConnectPlanner
    participant SP as SparkSession (classic)
    participant E as Spark engine

    C->>S: ExecutePlan(Plan)
    S->>P: transform(plan.proto)
    P->>SP: Dataset.ofRows(LogicalPlan)
    SP->>E: action (DAGScheduler.submitJob)
    E-->>SP: Iterator[InternalRow]
    SP-->>S: Arrow batches
    S-->>C: Stream of ExecutePlanResponse (Arrow + metrics + observed metrics)

Results are returned as Arrow batches over a server-streaming RPC, which matches how pyspark.sql.connect.dataframe.DataFrame.toPandas consumes them.

Protocol surface

The protobuf tree mirrors Catalyst's LogicalPlan and Expression hierarchies:

• Relation (in relations.proto) has oneof fields for Project, Filter, Join, Aggregate, Read (data source), WithColumns, Sort, Limit, SetOperation, etc.
• Expression (in expressions.proto) covers literals, column references, function calls, window specs, window expressions, lambda functions, and user-defined functions.
• Command (in commands.proto) covers SQL DDL and side-effect commands (e.g., RegisterFunction, WriteOperation, MergeIntoTable).
• Catalog (in catalog.proto) covers catalog inspection.
• MlCommand / MlRelation (in ml.proto) cover MLlib pipelines.
• Pipelines (in pipelines.proto) covers declarative pipelines.

The protocol is versioned by Spark version - the shims/ directory carries compatibility shims so older clients can target newer servers.

Authentication and sessions

• Sessions are identified by a session id and a user id in the gRPC metadata.
• The default deployment uses unauthenticated localhost; production deployments enable TLS and pluggable authentication via gRPC interceptors. See docs/spark-connect-overview.md and sql/connect/server/.../config/.
• Per-session state - SQLConf, temp views, temp functions, registered artifacts - is held by the SessionHolder (sql/connect/server/.../service/SessionHolder.scala).

Artifacts

A Connect client may upload jars, Python files, archives, or class files to add to the session classpath. The transport is in sql/connect/common/.../artifact/ and the server-side storage in sql/core/.../artifact/. On the engine side, each task picks the right artifact set via JobArtifactSet (core/.../JobArtifactSet.scala).

Python client highlights

• python/pyspark/sql/connect/session.py - SparkSession.builder.remote("sc://host:15002").
• python/pyspark/sql/connect/dataframe.py - the DataFrame mirror. Most methods build a Plan lazily and only materialize it on actions.
• python/pyspark/sql/connect/proto/ - generated protobuf bindings.
• python/pyspark/ml/connect/ - ML pipelines over Connect.

PySpark switches between classic and Connect modes via pyspark.sql.utils.is_remote() and a session factory in python/pyspark/sql/session.py.

Integration points

• Embeds a SparkSession per client; shares the underlying SparkContext.
• Uses the Catalyst analyzer and optimizer via the SparkConnectPlanner translation layer.
• Sends results in Arrow IPC format using the same code path as Dataset.toArrow.
• Listens on the same LiveListenerBus to expose observed metrics back to the client through Observation messages.

Entry points for modification

• Adding a new Relation operator: edit sql/connect/common/.../protobuf/spark/connect/relations.proto, regenerate stubs, then handle it in SparkConnectPlanner.transformRelation.
• Adding a new command: edit commands.proto and add a case in SparkConnectPlanner.transformCommand.
• Adding a Python client method: edit the corresponding file under python/pyspark/sql/connect/. The classic API in python/pyspark/sql/dataframe.py is the source-of-truth shape.
• Adding authentication: implement a gRPC ServerInterceptor and register it via spark.connect.grpc.interceptor.classes.