PySpark 装饰器¶

包裹在 @task.pyspark 装饰器中的 Python 可调用对象，如果可用，会被注入一个 SparkSession 和 SparkContext 对象。

参数¶

以下参数可以传递给装饰器

conn_id: str: 用于连接 Spark 集群的连接 ID。如果未指定，spark master 将设置为 local[*]。
config_kwargs: dict: 用于初始化 SparkConf 对象的 kwargs。这将覆盖连接中设置的 Spark 配置选项。

示例¶

以下示例展示了如何使用 @task.pyspark 装饰器。注意 spark 和 sc 对象会被注入到函数中。

tests/system/apache/spark/example_pyspark.py

@task.pyspark(conn_id="spark-local")
def spark_task(spark: SparkSession, sc: SparkContext) -> pd.DataFrame:
    df = spark.createDataFrame(
        [
            (1, "John Doe", 21),
            (2, "Jane Doe", 22),
            (3, "Joe Bloggs", 23),
        ],
        ["id", "name", "age"],
    )
    df.show()

    return df.toPandas()

Spark Connect¶

在 Apache Spark 3.4 中，Spark Connect 引入了一种解耦的客户端-服务器架构，允许使用 DataFrame API 远程连接到 Spark 集群。在 Airflow 中使用 Spark Connect 是利用 PySpark 装饰器的首选方式，因为它不需要在与 Airflow 相同的宿主机上运行 Spark 驱动程序。要使用 Spark Connect，请在您的主机 URL 前加上 sc://。例如，sc://spark-cluster:15002。

身份验证¶

Spark Connect 没有内置的身份验证。但是，gRPC HTTP/2 接口允许通过身份验证代理使用身份验证来与 Spark Connect 服务器通信。要使用身份验证，请确保创建一个 Spark Connect 连接并设置正确的凭据。