创建Spark会话时，您可以指定要使用Spark Connect，有几种方式可以做到这一点，如下所述。

如果您不使用这里概述的机制，您的Spark会话将像以前一样工作，而不利用Spark Connect。

设置 SPARK_REMOTE 环境变量

如果您在运行 Spark 客户端应用程序的客户端机器上设置 SPARK_REMOTE 环境变量，并创建一个新的 Spark 会话，如以下示例所示，则该会话将成为一个 Spark Connect 会话。使用这种方法，不需要更改代码即可开始使用 Spark Connect。

在终端窗口中，将 SPARK_REMOTE 环境变量设置为指向您之前在计算机上启动的本地Spark服务器：

export SPARK_REMOTE="sc://localhost"

并像往常一样启动Spark shell：

./bin/pyspark

PySpark shell 现在通过 Spark Connect 连接到 Spark，如欢迎消息所示：

Client connected to the Spark Connect server at localhost

在创建 Spark 会话时指定 Spark 连接

您还可以在创建 Spark 会话时明确指定要使用 Spark Connect。

例如，您可以启动带有Spark Connect的PySpark外壳，如此处所示。

要使用 Spark Connect 启动 PySpark shell，只需包含 remote 参数并指定 Spark 服务器的位置。在这个例子中，我们使用 localhost 连接到之前启动的本地 Spark 服务器：

./bin/pyspark --remote "sc://localhost"

你会注意到，PySpark shell 欢迎信息告诉你，你已通过 Spark Connect 连接到 Spark：

Client connected to the Spark Connect server at localhost

您还可以检查 Spark 会话类型。如果它包含 .connect. ，您正在使用 Spark Connect，如下面的示例所示：

SparkSession available as 'spark'.
>>> type(spark)
<class 'pyspark.sql.connect.session.SparkSession'>

现在你可以在命令行中运行PySpark代码，以查看Spark Connect的实际效果：

>>> columns = ["id","name"]
>>> data = [(1,"Sarah"),(2,"Maria")]
>>> df = spark.createDataFrame(data).toDF(*columns)
>>> df.show()
+---+-----+
| id| name|
+---+-----+
| 1|Sarah|
| 2|Maria|
+---+-----+

对于Scala shell，我们使用基于Ammonite的REPL，该REPL目前不包含在Apache Spark包中。

要设置新的 Scala shell，首先下载并安装 Coursier CLI 。然后，在终端窗口中使用以下命令安装 REPL：

cs install –-contrib spark-connect-repl

现在，您可以像这样启动基于 Ammonite 的 Scala REPL/外壳，以连接到您的 Spark 服务器：

spark-connect-repl

当 REPL 成功初始化时，将出现一条问候消息：

Spark session available as 'spark'.
   _____                  __      ______                            __
  / ___/____  ____ ______/ /__   / ____/___  ____  ____  ___  _____/ /_
  \__ \/ __ \/ __ `/ ___/ //_/  / /   / __ \/ __ \/ __ \/ _ \/ ___/ __/
 ___/ / /_/ / /_/ / /  / ,<    / /___/ /_/ / / / / / / /  __/ /__/ /_
/____/ .___/\__,_/_/  /_/|_|   \____/\____/_/ /_/_/ /_/\___/\___/\__/
    /_/

默认情况下，REPL 将尝试连接到本地 Spark 服务器。
在 shell 中运行以下 Scala 代码以查看 Spark Connect 的实际效果：

@ spark.range(10).count
res0: Long = 10L

配置客户端-服务器连接

默认情况下，REPL 将尝试连接到本地 Spark 服务器，端口为 15002。 但是，连接可以通过多种方式进行配置，如此配置 参考 所述。

设置 SPARK_REMOTE 环境变量

可以在客户端机器上设置SPARK_REMOTE环境变量，以自定义在REPL启动时初始化的客户端-服务器连接。

export SPARK_REMOTE="sc://myhost.com:443/;token=ABCDEFG"
spark-connect-repl

或者

SPARK_REMOTE="sc://myhost.com:443/;token=ABCDEFG" spark-connect-repl

使用CLI参数

自定义设置也可以通过以下所示的CLI参数传递：

spark-connect-repl --host myhost.com --port 443 --token ABCDEFG

支持的CLI参数列表可以在 此处 找到。

通过连接字符串进行程序化配置

连接也可以通过编程方式使用 SparkSession#builder 创建，如下例所示：

@ import org.apache.spark.sql.SparkSession
@ val spark = SparkSession.builder.remote("sc://localhost:443/;token=ABCDEFG").build()

首先，使用 pip install pyspark[connect]==3.5.0 安装 PySpark，或者如果构建一个打包的 PySpark 应用程序/库，请将其添加到你的 setup.py 文件中，如下所示：

install_requires=[
'pyspark[connect]==3.5.0'
]

在编写自己代码时，创建Spark会话时，请包含带有对您的Spark服务器引用的 remote 函数，如以下示例所示：

from pyspark.sql import SparkSession
spark = SparkSession.builder.remote("sc://localhost").getOrCreate()

为了说明，我们将创建一个简单的 Spark Connect 应用程序 SimpleApp.py：

"""SimpleApp.py"""
from pyspark.sql import SparkSession
logFile = "YOUR_SPARK_HOME/README.md" # 应该是你系统上的某个文件
spark = SparkSession.builder.remote("sc://localhost").appName("SimpleApp").getOrCreate()
logData = spark.read.text(logFile).cache()
numAs = logData.filter(logData.value.contains('a')).count()
numBs = logData.filter(logData.value.contains('b')).count()
print("Lines with a: %i, lines with b: %i" % (numAs, numBs))
spark.stop()

该程序仅计算文本文件中包含'a'的行数和包含'b'的行数。 注意，您需要将YOUR_SPARK_HOME替换为Spark安装的位置。

我们可以使用常规的Python解释器运行这个应用程序，如下所示：

# 使用Python解释器运行你的应用程序
$ python SimpleApp.py
...
行 with a: 72, 行 with b: 39

要在Scala应用程序/项目中使用Spark Connect，我们首先需要包含正确的依赖项。
以 sbt 构建系统为例，我们将以下依赖项添加到 build.sbt 文件中：

libraryDependencies += "org.apache.spark" %% "spark-sql-api" % "3.5.0"
libraryDependencies += "org.apache.spark" %% "spark-connect-client-jvm" % "3.5.0"

在编写自己代码时，创建Spark会话时，请包含带有对您的Spark服务器引用的 remote 函数，如以下示例所示：

import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder().remote("sc://localhost").build()

注意 : 参考用户定义代码的操作，如 UDF、过滤、映射等，需要注册一个 ClassFinder 以便拾取和上传任何所需的类文件。此外，任何 JAR 依赖项必须使用 SparkSession#AddArtifact 上传到服务器。

示例：

import org.apache.spark.sql.connect.client.REPLClassDirMonitor
// 注册一个 ClassFinder 以监控并上传构建输出的类文件。
val classFinder = new REPLClassDirMonitor(<ABSOLUTE_PATH_TO_BUILD_OUTPUT_DIR>)
spark.registerClassFinder(classfinder)
// 上传 JAR 依赖
spark.addArtifact(<ABSOLUTE_PATH_JAR_DEP>)

在这里， ABSOLUTE_PATH_TO_BUILD_OUTPUT_DIR 是构建系统写入类文件的输出目录， ABSOLUTE_PATH_JAR_DEP 是本地文件系统中JAR的位置。

提供的 REPLClassDirMonitor 是对 ClassFinder 的一个实现，它监视一个特定的目录，但您可以实现自己的类，扩展 ClassFinder 以进行自定义搜索和监视。