Managed Service for Apache Spark Connect 客户端 (google-cloud-spark-connect) 是一个基于开源 Apache Spark Connect 客户端构建的 Python 客户端库。它可让您的本地应用、脚本和笔记本通过 Spark Connect 协议在远程 Managed Service for Apache Spark 会话中运行 PySpark 工作负载。
与直接使用标准 Apache Spark Connect 客户端相比，google-cloud-spark-connect 具有以下优势：
- 自动化的会话生命周期和身份验证：
ManagedSparkSession.builder.getOrCreate()会自动预配、配置或重复使用远程 Managed Service for Apache Spark 交互式会话，并附加刷新的 Google Cloud 凭证。使用ManagedSparkSession编写的代码在作为 Managed Service for Apache Spark 批量工作负载提交时，也可以不加修改地运行，前提是批量工作负载的 Python 环境中安装了google-cloud-spark-connect软件包。 - 实时进度条和监控链接：在会话创建和 Spark 作业执行期间，在笔记本和交互式终端中显示实时进度条（跟踪各个阶段的任务完成情况），以及指向 Google Cloud 控制台中的会话详情页面、Spark 界面和各个 Spark 查询执行页面的直接链接。
- 交互式笔记本额外功能：安装
interactive额外功能后，在 IPython 或 Jupyter 笔记本中导入google.cloud.managed_spark_connect会自动加载：explore_dataframe()（由google-colabsqlviz提供支持），用于以交互方式探索、过滤、排序、聚合 PySpark DataFrame 并为其绘制图表，而无需编写绘图代码。%%sparksql和%sparksql魔术命令，用于在笔记本单元中运行 Spark SQL 查询，并提供交互式结果可视化和 Python 变量或 DataFrame 绑定。%dpip魔法命令，直接在远程会话的驱动程序和执行程序上安装 PyPI 软件包，而无需构建自定义容器映像。
与开源 Spark Connect 客户端一样，google-cloud-spark-connect 不需要本地 Spark 安装。使用 [client] extra 时，它只会安装轻量级 pyspark-client 软件包，而不是完整的 pyspark 发行版，因此您无需在机器上下载 Spark 或安装 Java 即可在 Managed Service for Apache Spark 上运行 PySpark。
本文档介绍了如何安装、配置和使用客户端。
准备工作
确保您拥有包含管理交互式会话和运行时配置文件（会话模板）所需权限的 Identity and Access Management 角色。
如果您在 Google Cloud之外运行客户端，请提供身份验证凭证。将
GOOGLE_APPLICATION_CREDENTIALS环境变量设置为服务账号密钥文件的路径。
安装或卸载客户端
您可以使用 pip 安装或卸载 google-cloud-spark-connect 软件包。
安装
对于本地笔记本和开发环境（例如，带有 Google Cloud Data Agent Kit 的 VS Code 或 JupyterLab），请在虚拟环境中通过命令行安装客户端：
pip install 'google-cloud-spark-connect[interactive,client]'
此命令会安装 google-cloud-spark-connect，并提供两个可选的额外功能：
interactive：安装交互式笔记本扩展程序（explore_dataframe()、%%sparksql和%dpip）。对于笔记本之外的独立 Python 脚本，可以省略此步骤。client：安装轻量级pyspark-client软件包，该软件包不需要本地 JVM。
在 Colab Enterprise 等代管式笔记本环境（其中预安装了完整的 pyspark）中，在笔记本顶部的专用设置单元中运行带有 spark 额外项的 %pip install，而不是 client：
%pip install 'google-cloud-spark-connect[interactive,spark]'
卸载
如需卸载客户端，请运行以下命令：
pip uninstall google-cloud-spark-connect pyspark-client
配置客户端
为会话指定项目和区域。您可以使用环境变量或在代码中使用 builder API 来设置这些值。
环境变量
设置 GOOGLE_CLOUD_PROJECT 和 GOOGLE_CLOUD_REGION 环境变量。
Builder API
使用 .projectId()、.location() 和 .sessionId() 方法。
spark = (
ManagedSparkSession.builder
.projectId("my-project")
.location("us-central1")
.sessionId("my-spark-session")
.getOrCreate()
)
启动 Spark 会话
如需启动 Spark 会话，请在 PySpark 应用或笔记本中导入 ManagedSparkSession，然后调用 ManagedSparkSession.builder.getOrCreate() API。我们建议您使用 .sessionId() 设置自定义会话 ID，以便在会话的存留时间 (TTL) 内重新运行笔记本或运行多个笔记本时，重复使用活跃会话并消除会话启动时间。
导入
ManagedSparkSession类。在 IPython 或 Jupyter 笔记本中，如果安装了interactiveextra，导入google.cloud.managed_spark_connect也会自动加载交互式笔记本 extra（explore_dataframe()、%%sparksql和%dpip）。调用
.sessionId()和.getOrCreate()以启动或重复使用会话。from google.cloud.managed_spark_connect import ManagedSparkSession spark = ManagedSparkSession.builder.sessionId("my-spark-session").getOrCreate()
新会话启动时，客户端会显示实时进度条以及指向 Google Cloud 控制台中会话详情页面的链接（在 VS Code 中，该链接会在 Data Agent Kit 中打开会话）。如果具有指定 ID 的活跃会话已存在，客户端会连接到该会话，而不是创建新会话。在这种情况下，客户端会忽略代码中的会话配置。如需应用配置更改，请终止现有会话 (spark.stop(terminate=True)) 或使用其他会话 ID。如果具有指定 ID 的会话已终止，客户端会自动删除旧会话，并使用相同 ID 创建新会话。会话处于活跃状态后：
- 在笔记本单元中评估
spark会显示指向会话详情页面和 Spark 界面的直接链接。 - 运行 Spark 操作（例如
df.show()、df.collect()或df.toPandas()）会显示实时任务进度条和指向 Spark 界面中操作执行页面的 Spark 查询链接。
配置 Spark 属性
如需配置 Spark 属性，请将一个或多个 .config() 方法链接到构建器。
from google.cloud.managed_spark_connect import ManagedSparkSession
spark = (
ManagedSparkSession.builder
.sessionId('my-spark-session')
.config('spark.executor.memory', '48g')
.config('spark.executor.cores', '8')
.getOrCreate()
)
使用高级配置
您可以直接使用构建器方法配置常见会话设置，例如会话 ID、运行时版本、子网、服务账号、会话模板、TTL（.ttl() / .ttlSeconds() 和 .idleTtl() / .idleTtlSeconds()）或标签：
import datetime
from google.cloud.managed_spark_connect import ManagedSparkSession
spark = (
ManagedSparkSession.builder
.projectId('my-project')
.location('us-central1')
.sessionId('my-spark-session')
.runtimeVersion('3.0')
.subnetwork('SUBNET')
.idleTtl(datetime.timedelta(minutes=30))
.getOrCreate()
)
如需完全自定义底层会话资源，请将 google.cloud.dataproc_v1.Session 对象传递给 .sessionConfig()：
from google.cloud.managed_spark_connect import ManagedSparkSession
from google.cloud.dataproc_v1 import Session
session_config = Session()
session_config.environment_config.execution_config.subnetwork_uri = 'SUBNET'
session_config.runtime_config.version = '3.0'
spark = (
ManagedSparkSession.builder
.projectId('my-project')
.location('us-central1')
.sessionId('my-spark-session')
.sessionConfig(session_config)
.getOrCreate()
)
重复使用已命名的会话
通过使用 .sessionId() 设置自定义会话 ID，您可以在多个笔记本中或在会话 TTL 内的内核重启期间重复使用单个 Spark 会话，从而消除会话启动延迟并降低费用。
在第一个笔记本中，创建一个具有自定义 ID 的会话。
from google.cloud.managed_spark_connect import ManagedSparkSession session_id = 'my-ml-pipeline-session' spark = ManagedSparkSession.builder.sessionId(session_id).getOrCreate() df = spark.createDataFrame([(1, 'data')], ['id', 'value']) df.show()在另一个笔记本中（或重启内核后），通过指定相同的会话 ID 来重复使用该会话。
from google.cloud.managed_spark_connect import ManagedSparkSession session_id = 'my-ml-pipeline-session' spark = ManagedSparkSession.builder.sessionId(session_id).getOrCreate() df = spark.createDataFrame([(2, 'more-data')], ['id', 'value']) df.show()
会话 ID 的长度必须介于 4 到 63 个字符之间，以小写字母开头，并且只能包含小写字母、数字和连字符。ID 不能以连字符结尾。命名会话会一直保留，直到被明确终止或达到其配置的 TTL。如果具有指定 ID 的会话已处于 TERMINATED、TERMINATING 或 FAILED 状态，则客户端会发出警告，删除旧会话，并创建具有相同 ID 的新会话（删除旧会话会从 Managed Service for Apache Spark 中移除其先前的元数据和 Spark 界面历史记录，但日志仍可在 Cloud Logging 中查看）。
如果已存在具有指定 ID 的活跃会话，则客户端会连接到该会话，并忽略代码中的会话配置。如需应用配置更改，请终止现有会话或使用其他会话 ID。
如需终止已命名的会话，请调用 spark.stop(terminate=True)。在不带参数的情况下调用 spark.stop() 会断开客户端连接，但会使命名会话保持运行。在没有 .sessionId() 的情况下创建的会话会被 spark.stop() 终止。
使用 Spark SQL 魔法命令
interactive extra 包含 %%sparksql 单元格魔法和 %sparksql 行魔法，用于在 Jupyter 笔记本中执行 Spark SQL 查询，并使用 google-colabsqlviz（explore_dataframe() 使用的同一可视化 widget）呈现交互式表格和图表。
当您使用 interactive extra 安装 google-cloud-spark-connect 并在笔记本中导入 google.cloud.managed_spark_connect 时，魔术扩展程序会自动加载。
可选：如果您尚未导入
google.cloud.managed_spark_connect，请明确加载 magic 扩展程序。%load_ext google.cloud.managed_spark_magics可选：配置默认设置。
%config SparkSql.limit=20使用
%%sparksql单元魔法命令（默认情况下会呈现交互式表格并将完整的 DataFrame 存储在_df中）或%sparksql行魔法命令（会抑制表格显示并直接返回 DataFrame）执行 SQL 查询：%%sparksql SELECT * FROM your_tabledf = %sparksql SELECT * FROM your_table可选：将 Python 变量和 DataFrame 传递到查询中。类型化 SQL 字面量值可以使用
:name绑定，表或列标识符可以使用IDENTIFIER(:name)绑定，笔记本命名空间中的 DataFrame 可以使用{df}绑定：min_id = 10 df = spark.createDataFrame([(5, 'a'), (15, 'b')], ['id', 'val'])%%sparksql SELECT * FROM {df} WHERE id >= :min_id
如需使用高级选项，请向 %%sparksql 命令添加标志。例如，如需缓存结果并创建视图，请运行以下命令：
%%sparksql --cache --view result_view df
SELECT * FROM your_table WHERE condition = true
您可以使用以下选项：
--cache或-c：缓存 DataFrame。--eager或-e：使用localCheckpoint(eager=True)立即计算结果。--view VIEW或-v VIEW：创建临时视图。--limit N或-l N：替换默认行显示限制（仅限单元格魔法）。--display MODE或-d MODE：设置显示模式（colabsqlviz或html；仅限单元格魔法）。--quiet或-q：禁止显示表。--explain [MODE]：输出查询计划，而不是显示结果。--session NAME或-s NAME：针对变量NAME中的SparkSession运行。variable_name：将结果存储在变量中（仅限单元魔法；默认为_df）。
在远程会话中安装 PyPI 软件包
如需在 PySpark 用户定义的函数 (UDF) 中使用第三方 Python 软件包，而无需构建自定义容器映像或重启会话，您可以直接在远程会话的驱动程序和执行程序上安装 PyPI 软件包：
Notebook 魔法命令
在具有活跃 ManagedSparkSession 的交互式笔记本中，使用 %dpip 魔术命令（在安装 interactive extra 时自动加载 google.cloud.managed_spark_connect）：
%dpip install scipy
%dpip 仅在远程会话中安装软件包。如需在 Spark 之外的本地代码中导入软件包，请在本地 Python 环境中也安装该软件包。
Python API
在脚本或笔记本中，将 pypi=True 传递给 spark.addArtifacts()：
spark.addArtifacts("scipy", pypi=True)
后续步骤
了解如何使用 Data Agent Kit 在 VS Code 中开发 Spark 代码。