Skip to content

[KYUUBI #7511][SPARK] Start Spark Connect in the Spark SQL engine #7670

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alexandrefimov wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[KYUUBI #7511][SPARK] Start Spark Connect in the Spark SQL engine #7670

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/configuration/settings.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
| kyuubi.engine.share.level.sub.domain | <undefined> | (deprecated) - Using kyuubi.engine.share.level.subdomain instead | string | 1.2.0 |
| kyuubi.engine.share.level.subdomain | <undefined> | Allow end-users to create a subdomain for the share level of an engine. A subdomain is a case-insensitive string values that must be a valid zookeeper subpath. For example, for the `USER` share level, an end-user can share a certain engine within a subdomain, not for all of its clients. End-users are free to create multiple engines in the `USER` share level. When disable engine pool, use 'default' if absent. | string | 1.4.0 |
| kyuubi.engine.single.spark.session | false | When set to true, this engine is running in a single session mode. All the JDBC/ODBC connections share the temporary views, function registries, SQL configuration and the current database. | boolean | 1.3.0 |
| kyuubi.engine.spark.connect.enabled | false | When true, the Spark SQL engine starts a Spark Connect service next to its Thrift frontend, so that Spark Connect clients can use the same engine. It requires a Spark distribution whose Spark Connect supports authentication, which is Spark 4.0 and later, a pre-shared `spark.connect.authenticate.token`, and the USER share level; the engine fails to start otherwise. | boolean | 1.13.0 |
| kyuubi.engine.spark.event.loggers | SPARK | A comma-separated list of engine loggers, where engine/session/operation etc events go.<ul> <li>SPARK: the events will be written to the Spark listener bus.</li> <li>JSON: the events will be written to the location of kyuubi.engine.event.json.log.path</li> <li>JDBC: to be done</li> <li>CUSTOM: to be done.</li></ul> | seq | 1.7.0 |
| kyuubi.engine.spark.initialize.sql | SHOW DATABASES | The initialize sql for Spark engine. It fallback to `kyuubi.engine.initialize.sql`. | seq | 1.8.1 |
| kyuubi.engine.spark.operation.incremental.collect | false | When true, the result will be sequentially calculated and returned to the Spark driver. Note that, kyuubi.operation.result.max.rows will be ignored on incremental collect mode. It fallback to `kyuubi.operation.incremental.collect` | boolean | 1.10.0 |
Expand Down
1 change: 1 addition & 0 deletions docs/deployment/spark/index.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,3 +30,4 @@ Even if you don't use Kyuubi, as a simple Spark user, I'm sure you'll find the n
dynamic_allocation
aqe
large_query_results
spark_connect
78 changes: 78 additions & 0 deletions docs/deployment/spark/spark_connect.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
-->

# Spark Connect in the Spark SQL Engine

The Spark SQL engine can start a Spark Connect service next to its Thrift frontend, so that
Spark Connect clients reach a Kyuubi-managed engine: the same `SparkContext`, the same session
extensions and so the same authorization rules a JDBC client gets. Spark gives each Connect session
its own `SparkSession`, so temporary views, registered functions and SQL configs are not shared with
the JDBC sessions of that engine.

The feature is off by default and is enabled per engine:

```properties
kyuubi.engine.spark.connect.enabled=true
spark.connect.authenticate.token=<a token this deployment generates>
```

`kyuubi.engine.spark.connect.enabled` is immutable and the server decides it per engine, so a
session cannot turn Spark Connect on by itself — neither under that key nor under the
`spark.kyuubi.` prefixed copy that the engine would otherwise read.

## Requirements

| Requirement | Why |
|----------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Spark 4.0 or later, with the `spark-connect` jar in the distribution | `spark.connect.authenticate.token` was added in Spark 4.0. Spark 3.5 serves Spark Connect without authentication, so the engine refuses to start rather than open an endpoint that authenticates no one |
| `spark.connect.authenticate.token`, or the `SPARK_CONNECT_AUTHENTICATE_TOKEN` environment variable | Every client that reaches the engine is authenticated by this pre-shared token |
| `kyuubi.engine.share.level=USER` and `kyuubi.engine.doAs.enabled=true` | A plan submitted over Spark Connect runs as the user the engine runs as, not as the `user_id` the client sends. Only these two together make that the session user; an engine started any other way serves Thrift only and says so in its launch log |

A runtime that cannot authenticate Spark Connect clients, or a missing token, fails the engine with
a message naming what is missing: the deployment asked for Connect and cannot have it. The share
level is the session's own choice, so an engine that would not run as the session user starts
without Connect instead of failing.

## What the engine configures

Unless the deployment sets them, the engine adds:

- `spark.plugins` gains `org.apache.spark.sql.connect.SparkConnectPlugin`, which is how Spark
starts Connect inside the driver;
- `spark.connect.grpc.binding.port=0`, so that engines of different users on the same host do not
compete for a fixed port. The port that was bound is logged as `sc://<driver host>:<port>`.

Configs the deployment sets are kept, so a fixed port or extra plugins keep working.

## Limitations

- The Spark Connect endpoint is not advertised in engine discovery yet, so clients need the
address from the engine log.
- Engine idle timeout and graceful shutdown still count Thrift sessions only. An engine whose only
client is a Spark Connect session can be terminated as idle.
- The token is passed to the engine like any other Spark config, so it is visible in the driver's
command line.
- On a Spark distribution without [SPARK-58658](https://issues.apache.org/jira/browse/SPARK-58658),
an authenticated Spark Connect client reads the engine's configuration back through the Config
RPC, and the secrets the server hands the engine are in it —
`spark.kyuubi.ha.zookeeper.auth.digest` among them. The engine names them in
`spark.redaction.regex`, but only the patched handler consults that pattern. The fix is merged
on every Spark branch and released in none: 4.0.4, 4.1.3 and 4.2.0 all predate it, so it arrives
in the next patch release of each line.
- Share levels other than `USER`, and engines started with `kyuubi.engine.doAs.enabled=false`, do
not serve Spark Connect.

11 changes: 11 additions & 0 deletions ...yuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ import org.apache.kyuubi.config.KyuubiConf._
import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_ENGINE_SUBMIT_TIME_KEY, KYUUBI_ENGINE_URL}
import org.apache.kyuubi.engine.ShareLevel
import org.apache.kyuubi.engine.spark.SparkSQLEngine.{countDownLatch, currentEngine}
import org.apache.kyuubi.engine.spark.connect.SparkConnectLauncher
import org.apache.kyuubi.engine.spark.events.{EngineEvent, EngineEventsStore, SparkEventHandlerRegister}
import org.apache.kyuubi.engine.spark.session.{SparkSessionImpl, SparkSQLSessionManager}
import org.apache.kyuubi.events.EventBus
Expand Down Expand Up @@ -326,6 +327,10 @@ object SparkSQLEngine extends Logging {
}

def createSpark(): SparkSession = {
if (SparkConnectLauncher.isEnabled(kyuubiConf)) {
SparkConnectLauncher.setup(kyuubiConf, _sparkConf)
}

val engineCredentials = kyuubiConf.getOption(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY)
kyuubiConf.unset(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY)
_sparkConf.set(s"spark.${KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY}", "")
Expand Down Expand Up @@ -407,6 +412,12 @@ object SparkSQLEngine extends Logging {
startInitTimeoutChecker(submitTime, initTimeout)
spark = createSpark()
sparkSessionCreated.set(true)
if (SparkConnectLauncher.isEnabled(kyuubiConf)) {
SparkConnectLauncher.boundEndpoint(spark.sparkContext.getConf).foreach {
case (host, port) =>
info(s"Spark Connect service is listening on sc://$host:$port")
}
}
try {
startEngine(spark)
// blocking main thread
Expand Down
Loading
Loading