[SPARK-43768][PYTHON][CONNECT] Python dependency management support in Python Spark Connect

[HyukjinKwon]

What changes were proposed in this pull request?

This PR proposes to add the support of archive (.zip, .jar, .tar.gz, .tgz, or .tar files) in SparkSession.addArtifacts so we can support Python dependency management in Python Spark Connect.

Why are the changes needed?

In order for end users to add the dependencies and archive files in Python Spark Connect client.

This PR enables the Python dependency management (https://www.databricks.com/blog/2020/12/22/how-to-manage-python-dependencies-in-pyspark.html) usecase in Spark Connect.

See below how to do this with Spark Connect Python client:

Precondition

Assume that we have a Spark Connect server already running, e.g., by:

./sbin/start-connect-server.sh --jars `ls connector/connect/server/target/**/spark-connect*SNAPSHOT.jar` --master "local-cluster[2,2,1024]"

and assume that you already have a dev env:

# Notice that you should install `conda-pack`.
conda create -y -n pyspark_conda_env -c conda-forge conda-pack python=3.9
conda activate pyspark_conda_env
pip install --upgrade -r dev/requirements.txt

Dependency management

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

import conda_pack
import os
# Pack the current environment ('pyspark_conda_env') to 'pyspark_conda_env.tar.gz'. 
# Or you can run 'conda pack' in your shell.
conda_pack.pack()  
spark.addArtifact(f"{os.environ.get('CONDA_DEFAULT_ENV')}.tar.gz#environment", archive=True)
spark.conf.set("spark.sql.execution.pyspark.python", "environment/bin/python")
# From now on, Python workers on executors use `pyspark_conda_env` Conda environment.

Run your Python UDFs

import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf("long")
def plug_one(s: pd.Series) -> pd.Series:
    return s + 1

spark.range(10).select(plug_one("id")).show()

Does this PR introduce any user-facing change?

Yes, it adds the support of archive (.zip, .jar, .tar.gz, .tgz, or .tar files) in SparkSession.addArtifacts.

How was this patch tested?

Manually tested as described above, and added a unittest.

Also, manually tested with local-cluster mode with the code below:

Also verified via:

import sys
from pyspark.sql.functions import udf

spark.range(1).select(udf(lambda x: sys.executable)("id")).show(truncate=False)

+----------------------------------------------------------------+
|<lambda>(id)                                                    |
+----------------------------------------------------------------+
|/.../spark/work/app-20230524132024-0000/1/environment/bin/python|
+----------------------------------------------------------------+

[HyukjinKwon]

This is actually pretty ugly workaround to support fragment in URI (but I believe this is the minimized change). Maybe we should pass URI instead of file path in Artifacts (?) but I would like to avoid touching the whole implementation in this PR. cc @hvanhovell @vicennial

[HyukjinKwon]

Merged to master.