El motor de cómputo Lindorm expone una API RESTful para enviar trabajos Python de Apache Spark. Úsela para ejecutar tareas de streaming y por lotes, así como tareas de aprendizaje automático y computación de grafos. Esta guía describe el flujo de trabajo completo: definición del trabajo, empaquetado de dependencias, carga de archivos a Object Storage Service (OSS) y envío del trabajo.
Requisitos previos
Antes de empezar, asegúrese de contar con lo siguiente:
Un motor de cómputo Lindorm activado. Consulte Activate the service
Un bucket de OSS para almacenar los archivos de su proyecto Python, el entorno de ejecución y el script de lanzamiento
Un AccessKey ID y un AccessKey secret con permisos de lectura y escritura sobre el bucket de OSS. Consulte Create an AccessKey pair
Un entorno Linux (obligatorio para empaquetar archivos binarios de Python compatibles con el motor de cómputo Lindorm)
Funcionamiento
El flujo de trabajo para enviar trabajos Python de Spark consta de cuatro etapas:
Definir el trabajo: estructure su proyecto Python con los archivos de punto de entrada necesarios.
Empaquetar el trabajo: agrupe el código del proyecto y el entorno de ejecución de Python por separado.
Cargar los archivos: almacene todos los artefactos en OSS.
Enviar el trabajo: ejecute el trabajo desde la consola de Lindorm o Data Management Service (DMS).
Paso 1: Definir el trabajo
Descargue el paquete de ejemplo de trabajo Spark y extráigalo. La carpeta extraída se llama lindorm-spark-examples. Revise el directorio lindorm-spark-examples/python para consultar la estructura de referencia del proyecto.
La raíz de su proyecto (your_project en el ejemplo) requiere tres cambios estructurales antes de poder enviarse.
1. Agregar __init__.py
Cree un archivo vacío llamado __init__.py en el directorio your_project. Esto convierte al directorio en un módulo de Python que launcher.py puede importar.
2. Preparar main.py
Abra your_project/main.py y realice dos modificaciones:
Agregue el directorio del proyecto a sys.path para que las importaciones se resuelvan correctamente durante la ejecución:
current_dir = os.path.abspath(os.path.dirname(__file__))
sys.path.append(current_dir)Envuelva su lógica de entrada en una función main(argv):
def main(argv):
# Write your job logic here
pass
if __name__ == "__main__":
main(sys.argv)Como ejemplo funcional, el siguiente código inicializa una SparkSession:
from pyspark.sql import SparkSession
def main(argv):
spark = SparkSession \
.builder \
.appName("PythonImportTest") \
.getOrCreate()
print(spark.conf)
spark.stop()
if __name__ == "__main__":
main(sys.argv)3. Crear launcher.py
En el directorio raíz de your_project, cree un archivo llamado launcher.py. Copie el contenido de lindorm-spark-examples/python/launcher.py. Este archivo es el punto de entrada que invoca el motor de cómputo Lindorm; agrega el directorio del proyecto a sys.path y llama a su función main(argv).
Paso 2: Empaquetar el trabajo
El empaquetado genera dos artefactos separados: un archivo .zip (o .egg) que contiene el código del proyecto, y un archivo tar que contiene el entorno de ejecución de Python.
Empaquetar el código del proyecto
Comprima your_project en un archivo .zip:
zip -r your_project.zip your_projectAlternativamente, cree un archivo .egg. Consulte Building Eggs.
Empaquetar el entorno de ejecución de Python
Utilice Conda o Virtualenv para empaquetar el entorno de ejecución de Python y las bibliotecas de terceros en un archivo tar; luego páselo mediante el parámetro spark.archives.
|
Herramienta |
Cuándo usarla |
|
Conda |
Úsela cuando su trabajo requiera una versión específica de Python o deba ejecutarse en nodos donde Python no está preinstalado. Conda incluye el intérprete de Python dentro del archivo tar. |
|
Virtualenv |
Úselo cuando la versión de Python en los nodos del clúster ya coincida con la de su proyecto. Virtualenv no incluye el intérprete, sino que utiliza la instalación de Python existente en el nodo. |
Ejecute el paso de empaquetado en Linux. El motor de cómputo Lindorm requiere archivos binarios de Python compilados para Linux.
Ejemplo: uso de Conda
conda create -y -n pyspark_conda_env -c conda-forge numpy conda-pack
conda activate pyspark_conda_env
conda pack -f -o pyspark_conda_env.tar.gzPara otras opciones de empaquetado, consulte Python Package Management.
Paso 3: Cargar los archivos a OSS
Cargue los tres artefactos a su bucket de OSS. Consulte Simple upload.
launcher.py: el punto de entrada creado en el paso 1your_project.zip(o.egg): el paquete de código del proyecto generado en el paso 2pyspark_conda_env.tar.gz: el entorno de ejecución de Python generado en el paso 2
Paso 4: Enviar el trabajo
El motor de cómputo Lindorm admite dos métodos de envío:
Consola de Lindorm: Consulte Manage jobs in the console.
DMS: Consulte Manage jobs using DMS.
Sin importar el método elegido, configure los siguientes parámetros en el campo configs de su solicitud de envío de trabajo.
Parámetros del entorno de ejecución
Establezca estos tres parámetros para dirigir el trabajo hacia sus artefactos en OSS:
|
Parámetro |
Descripción |
Ejemplo |
|
|
Ruta al archivo |
|
|
|
Ruta al archivo tar del entorno de ejecución de Python. Use |
|
|
|
Ruta al ejecutable de Python dentro del archivo tar extraído |
|
Parámetros de acceso a OSS
Configure estos parámetros para que el motor de cómputo pueda leer sus archivos desde OSS:
|
Parámetro |
Descripción |
Ejemplo |
|
|
Endpoint del bucket de OSS donde se almacenan sus archivos Python |
|
|
|
AccessKey ID |
|
|
|
AccessKey secret |
|
|
|
Clase utilizada para acceder a OSS |
|
Para más parámetros de OSS sobre Hadoop, consulte la documentación de Hadoop-Aliyun.
Ejemplo: valor de configs ensamblado
A continuación se muestran todos los parámetros del entorno de ejecución y de acceso a OSS combinados en un único objeto configs:
{
"spark.archives": "oss://testBucketName/pyspark_conda_env.tar.gz#environment",
"spark.kubernetes.driverEnv.PYSPARK_PYTHON": "./environment/bin/python",
"spark.submit.pyFiles": "oss://testBucketName/your_project.zip",
"spark.hadoop.fs.oss.endpoint": "oss-cn-beijing-internal.aliyuncs.com",
"spark.hadoop.fs.oss.accessKeyId": "<your-access-key-id>",
"spark.hadoop.fs.oss.accessKeySecret": "<your-access-key-secret>",
"spark.hadoop.fs.oss.impl": "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem"
}Reemplace <your-access-key-id> y <your-access-key-secret> con sus credenciales de AccessKey reales.
Diagnóstico de trabajos
Después de enviar el trabajo, consulte su estado y la dirección de la interfaz de usuario de Spark en la página Jobs. Consulte View a job.
Si el envío falla, envíe un ticket y proporcione el ID del trabajo y la dirección de la interfaz de usuario de Spark al equipo de soporte.