Develop Spark applications with PySpark - AnalyticDB - Centro de documentación de Alibaba Cloud

Este artículo describe cómo desarrollar trabajos de Python para Spark en AnalyticDB for MySQL y proporciona un método para configurar un entorno en la nube.

Requisitos previos

Se creó un clúster de AnalyticDB for MySQL Enterprise Edition, Basic Edition o Data Lakehouse Edition.
Se creó un bucket de Object Storage Service (OSS) en la misma región que el clúster de AnalyticDB for MySQL.
A job resource group is created para el clúster de AnalyticDB for MySQL Enterprise Edition, Basic Edition o Data Lakehouse Edition.
Se creó una cuenta de base de datos para el clúster de AnalyticDB for MySQL.
- Si utiliza una cuenta de Alibaba Cloud, solo necesita create a privileged account.
- Si utiliza un usuario de Resource Access Management (RAM), debe create a privileged account and a standard account y associate the standard account with the RAM user.

Uso básico de PySpark

Write the following sample code and save it as example.py.

from pyspark.sql import SparkSession
if __name__ == "__main__":
    spark = SparkSession.builder.getOrCreate()
    df = spark.sql("SELECT 1+1")
    df.printSchema()
    df.show()

Upload example.py to OSS. For more information, see Upload a file.
Go to the Spark development editor.
1. Inicie sesión en la consola de AnalyticDB for MySQL. En la esquina superior izquierda de la consola, seleccione una región. En el panel de navegación izquierdo, haga clic en Clusters. Busque el clúster que desea gestionar y haga clic en su ID.
2. In the left-side navigation pane, choose Job Development > Spark JAR Development.
At the top of the editor window, select a job resource group and a Spark job type. This topic uses the Batch type as an example.

Enter the following job configuration in the editor.

{
 "name": "Spark Python Test",
 "file": "oss://testBucketName/example.py",
 "conf": {
 "spark.driver.resourceSpec": "small",
 "spark.executor.instances": 1,
 "spark.executor.resourceSpec": "small"
 }
}

Para obtener más información sobre los parámetros, consulte Parameter description.

Uso de dependencias de Python

Este tema presenta dos soluciones para configurar un entorno en la nube sin necesidad de empaquetado local.

Tipo de solución

Casos de uso

Ventajas y desventajas

Instalación en tiempo real

Depuración y trabajos temporales.
Pocas dependencias y descarga rápida.

Ventajas: Configuración sencilla y listo para usar.

Desventajas: Las dependencias se descargan e instalan cada vez que se ejecuta el trabajo.

Compilación en la nube

Entornos de producción.
Muchas dependencias o dependencias grandes.
Entornos que requieren reutilización a largo plazo.

Ventajas: Empaquete una vez y reutilice indefinidamente. Garantiza un inicio rápido y alta estabilidad.

Desventajas: Requiere un trabajo adicional para empaquetar las dependencias.

Requisitos previos

Antes de configurar el entorno en la nube, asegúrese de cumplir con los siguientes requisitos:

Estado del clúster: El clúster está inicializado y puede ejecutar correctamente el ejemplo básico Spark Pi.
Requisitos de versión:
- Versión de Spark: Se admite la versión 3.5.1.
- Versión de Python: Se admiten las versiones 3.9 o 3.11.
Restricción clave (versión de NumPy):
- Apache Spark debe ejecutarse en un entorno con numpy < 2.0.0.
- El sistema impone la instalación de numpy==1.26.0.
- Importante: Asegúrese de que otras dependencias que instale, como Pandas y SciPy, sean compatibles con NumPy 1.26.0. De lo contrario, el trabajo fallará.

Ejemplos

Instalación en tiempo real

Prepare el código de su aplicación

Escriba un script de Python, por ejemplo, job.py, y cárguelo en una ruta de OSS, como oss://your-bucket/scripts/job.py.

# This sample script prints all dependencies in the current Python environment.
# Print all modules in the Python environment
import pkgutil
if __name__ == "__main__":
    for module_info in pkgutil.iter_modules():
        print(module_info.name)

Configure los parámetros del trabajo

## Sample job
{
    "file": "oss://your-bucket/scripts/job.py",  // Path to your code file
    "name": "Real-time Env Demo",
    "conf": {
        "spark.adb.version": "3.5",
        "spark.driver.resourceSpec": "medium",
        "spark.executor.instances": 1,
        "spark.executor.resourceSpec": "medium",
        // --- Start of core configuration ---
        // 1. Specify the Python version
        "spark.kubernetes.driverEnv.PYTHON_BIN": "python3.11",
        "spark.executorEnv.PYTHON_BIN": "python3.11",
        // 2. Specify the dependencies to install (for both Driver and Executor)
        "spark.kubernetes.driverEnv.PYTHON_MODULES": "chinesecalendar>=1.10.0,pandas>=1.5.3,lunar_python",
        "spark.executorEnv.PYTHON_MODULES": "chinesecalendar>=1.10.0,pandas>=1.5.3,lunar_python"
        // --- End of core configuration ---
    }
}

Importante

Spark consta de un Driver (nodo de control) y Executors (nodos de ejecución). Para garantizar un entorno consistente, debe configurar las mismas variables de entorno para ambos.

Parámetro	Descripción	Obligatorio	Predeterminado	Notas
spark.kubernetes.driverEnv.PYTHON_MODULES	Lista de paquetes de Python para instalar.	Sí	Ninguna	Separe varias dependencias de Python con comas (,). El formato de las dependencias de Python debe cumplir totalmente con los requisitos de la comunidad `PyPI`. Las dependencias de Python sin restricciones de versión deben colocarse al final de la lista. Ejemplo: `chinesecalendar>=1.10.0,dynaconf>=3.2.10,pandas>=1.5.3,lunar_python`
spark.executorEnv.PYTHON_MODULES
spark.kubernetes.driverEnv.PYTHON_BIN	Versión de Python que se utilizará para el trabajo.	No	python3.11	Valores válidos: python3.11 python3.9
spark.executorEnv.PYTHON_BIN
spark.kubernetes.driverEnv.INDEX_URL	URL del repositorio `PyPI`.	No	http://mirrors.cloud.aliyuncs.com/pypi/simple/	El valor predeterminado es la URL de un espejo alojado en Alibaba Cloud, accesible a través de la red interna. Si especifica una dirección accesible solo mediante la red pública, como el espejo PyPI de la Universidad Tsinghua, debe habilitar el acceso a la red pública. Para obtener más información, consulte Configure public network access for a Spark application.
spark.executorEnv.INDEX_URL
spark.kubernetes.driverEnv.TRUSTED_HOST	Dominio del repositorio `PyPI` que se agregará como host de confianza.	No	mirrors.cloud.aliyuncs.com	Python verifica el certificado SSL del repositorio PyPI durante la instalación. Si el certificado del repositorio no proviene de una autoridad de certificación (CA) de confianza, utilice este parámetro para marcar el dominio del repositorio como host de confianza. Importante Utilice este parámetro con precaución. Asegúrese de que la fuente PyPI configurada sea confiable, ya que los ataques de confusión de dependencias son una amenaza común.
spark.executorEnv.TRUSTED_HOST

Ejecute el trabajo de ejemplo. Puede consultar el registro para comprobar que el entorno de Python contiene las dependencias declaradas y sus dependencias transitivas.

xxlimited_35
zlib
numpy
pandas
_distutils_hack
_virtualenv
chinese_calendar
dateutil
lunar_python
pip
pkg_resources
pytz
setuptools
six
tzdata
wheel
>>>>>>>> stderr:
25/12/25 17:01:56 INFO ShutdownHookManager: Shutdown hook called

Compilación en la nube

Esta solución utiliza un trabajo dedicado para empaquetar las dependencias en un archivo comprimido, que luego se carga en OSS para su reutilización en trabajos posteriores.

Planifique una ruta de OSS

Especifique una ruta de OSS para almacenar el entorno empaquetado, por ejemplo, oss://your-bucket/envs/my_custom_env.

Envíe un trabajo de empaquetado

Importante

No modifique la ruta del script de empaquetado integrado: local:///opt/tools/build_venv.py.
En args, especifique las dependencias que se instalarán.

## Sample job
{
    // 1. Specify all dependencies to be packaged.
    "args": [
        "chinesecalendar>=1.10.0",
        "pandas>=1.5.3",
        "pyarrow>=19.0.1",
        "lunar_python"
    ],
    // 2. Call the built-in packaging script (do not modify).
    "file": "local:///opt/tools/build_venv.py",
    "name": "Build VirtualEnv Job",
    "conf": {
        "spark.driver.resourceSpec": "medium",
        "spark.executor.instances": 1,
        "spark.executor.resourceSpec": "medium",
        // 3. Specify the Python version.
        "spark.kubernetes.driverEnv.PYTHON_BIN": "python3.11",
        // 4. Specify the OSS path to upload the packaged environment (modify as needed).
        "spark.kubernetes.driverEnv.VENV_OSS_PATH": "oss://your-bucket/envs/my_custom_env",
        // 5. Specify the temporary directory for the build.
        "spark.kubernetes.driverEnv.VENV_DIR": "/tmp/build_test"
    }
}

Parámetro	Descripción	Obligatorio	Predeterminado	Notas
spark.kubernetes.driverEnv.VENV_OSS_PATH	Ruta de almacenamiento para el paquete del entorno.	Sí	Ninguna	Ejemplo: `oss://your-bucket/envs/my_custom_env`.
spark.kubernetes.driverEnv.VENV_DIR	Directorio temporal de compilación.	No	/tmp/venv	Si el paquete del entorno es grande, monte un disco de datos y cambie esta ruta a `/user_data_dir`.

spark.kubernetes.driverEnv.PYTHON_BIN	Versión de Python que se utilizará para el trabajo.	No	python3.11	Valores válidos: python3.11 python3.9
spark.kubernetes.driverEnv.INDEX_URL	URL del repositorio `PyPI`.	No	http://mirrors.cloud.aliyuncs.com/pypi/simple/	El valor predeterminado es la URL de un espejo alojado en Alibaba Cloud, accesible a través de la red interna. Si especifica una dirección accesible solo mediante la red pública, como el espejo PyPI de la Universidad Tsinghua, debe habilitar el acceso a la red pública. Para obtener más información, consulte Configure public network access for a Spark application.

spark.kubernetes.driverEnv.TRUSTED_HOST	Dominio del repositorio `PyPI` que se agregará como host de confianza.	No	mirrors.cloud.aliyuncs.com	Python verifica el certificado SSL del repositorio PyPI durante la instalación. Si el certificado del repositorio no proviene de una autoridad de certificación (CA) de confianza, utilice este parámetro para marcar el dominio del repositorio como host de confianza. Importante Utilice este parámetro con precaución. Asegúrese de que la fuente PyPI configurada sea confiable, ya que los ataques de confusión de dependencias son una amenaza común.

Ejecute el trabajo

Los registros muestran los detalles de la carga del archivo y una lista completa de los paquetes y sus versiones instalados en el entorno virtual.

------------------ --------------
chinesecalendar    1.11.0
lunar_python       1.4.8
numpy              1.26.0
pandas             2.3.3
pip                24.2
pyarrow            22.0.0
python-dateutil    2.9.0.post0
pytz               2025.2
setuptools         75.1.0
six                1.17.0
tzdata             2025.3
wheel              0.44.0
Uploading archive to oss://xxx/envs/my_custom_env/venv_20251225174458.tar.gz
+ ossutil cp /tmp/venv_20251225174458.tar.gz xxx
xxx
xxx
xxx
xxx
xxx
xxx
xxx
xxx
xxx
Succeed: Total num: 1, size: 115,655,687. OK num: 1(upload 1 files).
0.788886(s) elapsed
Upload completed: oss://xxx/envs/my_custom_env/venv_20251225174458.tar.gz

Utilice el paquete del entorno

Para trabajos posteriores de PySpark, puede hacer referencia al archivo en oss://your-bucket/envs/my_custom_env.

Para utilizar el entorno empaquetado, establezca el parámetro archives en la configuración de su trabajo.

Aviso

Esta configuración solo es compatible con Spark 3.5 y versiones posteriores.

## Sample usage
{
 "name": "Spark Python",
 "file": "oss://testBucketName/example.py",
 "archives": ["oss://your-bucket/envs/my_custom_env/venv_*****.tar.gz#PY3"],
 "args": [
 "oss://testBucketName/staff.csv"
 ],
 "conf": {
 "spark.driver.resourceSpec": "small",
 "spark.executor.instances": 2,
 "spark.executor.resourceSpec": "small",
 "spark.pyspark.python": "./PY3/venv/bin/python3"
 }
}

Solución de problemas

ModuleNotFoundError:
- Compruebe que las dependencias estén configuradas tanto para driverEnv como para executorEnv.
- Verifique que los nombres de los paquetes estén escritos correctamente y coincidan con los nombres en PyPI.
Errores relacionados con NumPy:

Compruebe si sus dependencias requieren numpy >= 2.0.0. De ser así, reduzca las versiones de sus dependencias para que sean compatibles con numpy 1.26.0.
Tiempos de espera de descarga agotados:

Si los tiempos de espera continúan ocurriendo al utilizar el espejo de red interna predeterminado, compruebe si ha especificado un espejo público y si también ha habilitado el acceso a la red pública para su trabajo.