When using Spark Connect to execute tasks, the code is actually run by the Executor of the remote Spark Connect App. If local custom libraries, auxiliary files, or archives are used in the task, these dependencies must be explicitly provided to the App so that the Executor can correctly load them at runtime.
Currently, two methods are supported for providing dependencies:

• Pre-configure them through parameters when creating the App in the console.
• Dynamically upload them at runtime using the addArtifact interface (Python only).

When dependency packages are large, it is recommended to configure them during App creation to avoid repeated uploads for each run.

For creation guidance, see Create in EMR console, and specify the dependency package path using the Archive parameter to upload dependencies.

The Spark Connect client supports uploading the Python environment and dependency files to the Spark Connect App via the addArtifact interface, which are then distributed to each Executor through SparkContext.

Python environment (Conda packaging recommended) #

Important:

1. Packaging must be performed in a Linux environment.
2. The Python versions of the Driver and Executor environments must be consistent. Serverless Spark uses Python 3.12 by default.
3. If the Conda package is large, uploading may take a long time. It is recommended to configure it when creating the Spark Connect App.

import os
from pathlib import Path
import conda_pack
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StringType

# Create SparkSession
spark = SparkSession.builder.getOrCreate()

@F.udf(StringType())
def probe(_: int) -> str:
    import sys
    return f"python={sys.executable}; version={sys.version.split()[0]}; prefix={sys.prefix}"


def show_probe(label: str) -> None:
    spark.range(1).select(probe(F.lit(1)).alias(label)).show(truncate=False)


show_probe("before_addartifact")

# Package the current environment's pyspark_conda_env as pyspark_conda_env.tar.gz. You can also use the conda pack command to package it.
# Important
# 1. Packaging must be performed in a Linux environment.
# 2. The Python versions of the Driver and Executor environments must be consistent.
conda_pack.pack()
spark.addArtifact(
    f"{os.environ.get('CONDA_DEFAULT_ENV')}.tar.gz#environment",
    archive=True)

# Set the executor Python path
spark.conf.set(
    "spark.sql.execution.pyspark.python", "environment/bin/python")

show_probe("after_addartifact")

File #

from pathlib import Path
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StringType

root = Path.cwd()
spark.addArtifact(str(root / "hello_connect.txt"), file=True)


@F.udf(StringType())
def read_file(_: int) -> str:
    from pyspark import SparkFiles

    with open(SparkFiles.get("hello_connect.txt"), "r", encoding="utf-8") as f:
        return f.read().strip()


spark.range(1).select(read_file(F.lit(1)).alias("file_result")).show(truncate=False)

Pyfile (supports Python files and zip archives) #

from pathlib import Path
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StringType

root = Path.cwd()
spark.addArtifact(str(root / "connect_demo_helper.py"), pyfile=True)
# Automatic extraction of zip archives is supported
spark.addArtifact(str(root / "demo_archive.zip"), pyfile=True)


@F.udf(StringType())
def use_pyfile(_: int) -> str:
    import connect_demo_helper

    return connect_demo_helper.describe_pyfile_artifact()


spark.range(1).select(use_pyfile(F.lit(1)).alias("pyfile_result")).show(truncate=False)

Archive #

from pathlib import Path
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StringType

root = Path.cwd()
spark.addArtifact(str(root / "demo_archive.zip"), archive=True)


@F.udf(StringType())
def read_archive(_: int) -> str:
    from pyspark import SparkFiles

    archive_dir = SparkFiles.get("demo_archive.zip")
    archive_file = os.path.join(archive_dir, "archive_message.txt")
    with open(archive_file, "r", encoding="utf-8") as f:
        return f.read().strip()


spark.range(1).select(read_archive(F.lit(1)).alias("archive_result")).show(truncate=False)