Create and use an EMR Spark node - DataWorks - Centro de documentación de Alibaba Cloud

Spark es un motor de análisis unificado para big data, reconocido por su alto rendimiento, facilidad de uso y versatilidad. Admite cómputo complejo en memoria, lo que lo hace ideal para crear aplicaciones de análisis de datos a gran escala y con baja latencia. DataWorks permite utilizar nodos EMR Spark para desarrollar y programar periódicamente tareas de Spark. Este tema explica cómo configurar y usar nodos EMR Spark, e incluye ejemplos de aplicación.

Requisitos previos

Antes de comenzar el desarrollo del nodo, si requiere un entorno de componentes personalizado, cree una imagen personalizada basada en la imagen oficial dataworks_emr_base_task_pod y utilícela en Data Studio. Para obtener más información, consulte Create a custom image y Use images in data development.

Por ejemplo, puede reemplazar paquetes JAR de Spark o agregar dependencias de libraries, files o JAR packages específicos al crear una imagen personalizada.
Cree un clúster de Alibaba Cloud EMR y regístrelo en DataWorks. Para obtener más información, consulte New Data Studio: Attach an EMR compute resource.
(Opcional, para usuarios de RAM) El usuario de Resource Access Management (RAM) destinado al desarrollo de tareas debe agregarse al espacio de trabajo y asignársele el rol Development o Workspace Administrator (este rol incluye permisos amplios y debe otorgarse con precaución). Para obtener más información, consulte Add workspace members.

Si utiliza una cuenta raíz, omita este paso.
Si sus tareas requieren un entorno de desarrollo específico, puede construir una imagen personalizada. Para obtener más información, consulte custom image.

Limitaciones

Los nodos EMR Shell solo pueden ejecutarse en un serverless resource group (recomendado) o en un resource group exclusivo para programación. El uso de una imagen personalizada para el desarrollo de datos requiere un resource group serverless.
Para gestionar metadatos de un clúster DataLake o personalizado en DataWorks, primero debe configurar EMR-HOOK en el clúster. Para obtener más información, consulte Configure EMR-HOOK for Spark SQL.

Nota
Sin EMR-HOOK configurado en el clúster, DataWorks no puede mostrar metadatos en tiempo real, generar registros de auditoría, mostrar linaje de datos ni realizar tareas de gobernanza relacionadas con EMR.
Los clústeres EMR on ACK Spark no admiten la visualización del linaje de datos, pero los clústeres EMR Serverless Spark sí lo permiten.
Los clústeres EMR on ACK Spark y EMR Serverless Spark solo admiten la referencia a recursos de OSS mediante OSS REF y la carga de recursos en OSS, pero no la carga de recursos en HDFS.
Los clústeres DataLake y los clústeres personalizados admiten la referencia a recursos de OSS mediante OSS REF, la carga de recursos en OSS y la carga de recursos en HDFS.

Notes

Si ha habilitado el control de acceso Ranger para Spark en el clúster EMR asociado al espacio de trabajo actual:

Las tareas de Spark que utilizan la imagen predeterminada admiten esta función automáticamente.
Para ejecutar una tarea de Spark con una imagen personalizada, envíe un ticket al soporte técnico para solicitar una actualización de la imagen.

Desarrollar un trabajo de Spark y obtener el paquete JAR

Antes de poder programar trabajos EMR Spark con DataWorks, debe desarrollar el trabajo en EMR y compilar el código en un paquete JAR. Para obtener más información sobre el desarrollo de trabajos EMR Spark, consulte Spark Overview.

Nota

Posteriormente, deberá cargar el paquete JAR en DataWorks para programar trabajos recurrentes de EMR Spark.

Procedimiento

En la página de edición del nodo EMR Spark, desarrolle el trabajo.

Desarrollar un trabajo de Spark

Elija una opción según su escenario:

Opción 1: Cargar y hacer referencia

DataWorks permite cargar recursos desde su máquina local a DataStudio y luego referenciarlos. Después de compilar el trabajo EMR Spark, debe obtener el paquete jar compilado. El método de almacenamiento recomendado para el paquete jar depende de su tamaño.

Cargue el paquete jar y créelo como un recurso EMR de DataWorks; luego envíelo. También puede almacenarlo directamente en HDFS en el clúster EMR. Los clústeres EMR on ACK y EMR Serverless Spark no admiten la carga de recursos en HDFS.

JAR < 500 MB

Cree un recurso JAR de EMR.

Si un paquete jar tiene un tamaño menor a 500 MB, puede cargarlo desde su máquina local como un recurso JAR de EMR en DataWorks. Esto facilita la gestión visual en la consola de DataWorks. Después de crear el recurso, envíelo. Para obtener más información, consulte Create and use EMR resources.
1. Cargue el paquete jar en el directorio de almacenamiento para recursos JAR mediante el método de carga Local. Para obtener más información, consulte Resource management.
2. Haga clic en Click Upload para cargar el recurso JAR.
3. Seleccione una Storage Path, Data Sources y un Resource Group.
4. Haga clic en Save.
Referencie el recurso JAR de EMR.
1. Abra el nodo EMR Spark creado y vaya a la página de edición de código.
2. En la sección Resource Management del panel de navegación izquierdo, busque el recurso que desea referenciar, haga clic derecho sobre él y seleccione Insert Resource Path.
3. Al seleccionar la referencia, se agrega automáticamente una instrucción de referencia al editor de código del nodo EMR Spark. Esto indica que la referencia se realizó correctamente.
```
##@resource_reference{"spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jar"}
spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jar
```
  En este código, spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jar es el nombre del recurso JAR de EMR que cargó.
4. Edite el código del nodo EMR Spark para incluir el comando spark-submit. El siguiente código es un ejemplo.
  
  Nota
  El editor de nodos EMR Spark no admite comentarios. Escriba el código del trabajo como se muestra en el siguiente ejemplo, sin comentarios. De lo contrario, el nodo no se ejecutará.
```
##@resource_reference{"spark-examples_2.11-2.4.0.jar"}
spark-submit --class org.apache.spark.examples.SparkPi --master yarn  spark-examples_2.11-2.4.0.jar 100
```
  Nota
  - org.apache.spark.examples.SparkPi: La clase principal en el paquete jar compilado.
  - spark-examples_2.11-2.4.0.jar: El nombre del recurso JAR de EMR que cargó.
  - Puede usar los valores de ejemplo para los demás parámetros. También puede ejecutar el comando spark-submit --help para ver la información de uso y modificar el comando spark-submit según sea necesario.
  - Para usar parámetros simplificados en el comando spark-submit dentro de un nodo Spark, debe agregarlos a su código; por ejemplo, --executor-memory 2G.
  - Los nodos Spark solo admiten el envío de trabajos a YARN en modo clúster.
  - Para los trabajos enviados con spark-submit, establezca el modo de despliegue en modo clúster, no en modo cliente.

JAR >= 500 MB

Cree un recurso JAR de EMR.

Si un paquete jar tiene un tamaño de 500 MB o superior, no puede cargarlo desde su máquina local como un recurso de DataWorks. Se recomienda almacenar el paquete jar directamente en HDFS en el clúster EMR y registrar su ruta de almacenamiento. Luego podrá referenciar esta ruta al programar el trabajo de Spark en DataWorks.
1. Cargue el paquete jar en el directorio de almacenamiento para recursos JAR mediante el método de carga Local. Para obtener más información, consulte Resource management.
2. Haga clic en Click Upload para cargar el recurso JAR.
3. Seleccione una Storage Path, Data Sources y un Resource Group.
4. Haga clic en Save.
Referencie el recurso JAR de EMR.

Si el paquete jar está almacenado en HDFS, puede referenciarlo especificando su ruta en el código del nodo EMR Spark.
1. Haga doble clic en el nodo EMR Spark creado para abrir su página de editor de código.
2. Escriba el comando spark-submit. A continuación se muestra un ejemplo.
```
spark-submit --master yarn
--deploy-mode cluster
--name SparkPi
--driver-memory 4G
--driver-cores 1
--num-executors 5
--executor-memory 4G
--executor-cores 1
--class org.apache.spark.examples.JavaSparkPi
hdfs:///tmp/jars/spark-examples_2.11-2.4.8.jar 100
```
  Nota
  - hdfs:///tmp/jars/spark-examples_2.11-2.4.8.jar: La ruta real del paquete jar en HDFS.
  - org.apache.spark.examples.JavaSparkPi: La clase principal en el paquete jar compilado.
  - Configure los demás parámetros según la configuración de su clúster EMR. También puede ejecutar el comando spark-submit --help para ver la información de uso de spark-submit y modificar el comando según sea necesario.
  - Para usar parámetros simplificados en el comando spark-submit dentro de un nodo Spark, debe agregarlos a su código; por ejemplo, --executor-memory 2G.
  - Los nodos Spark solo admiten el envío de trabajos a YARN en modo clúster.
  - Para los trabajos enviados con spark-submit, establezca el modo de despliegue en modo clúster, no en modo cliente.

Opción 2: Referenciar un recurso de OSS

Puede referenciar directamente recursos de OSS desde el nodo actual mediante OSS REF. Al ejecutar un nodo EMR, DataWorks carga automáticamente los recursos de OSS especificados en el código para su uso local. Este método se utiliza comúnmente en escenarios donde los trabajos de EMR requieren dependencias JAR o dependen de scripts.

Desarrolle el recurso JAR.

Prepare las dependencias de código.

Puede consultar las dependencias de código necesarias en la ruta /usr/lib/emr/spark-current/jars/ del nodo maestro de su clúster EMR. El siguiente ejemplo utiliza Spark 3.4.2. Debe abrir un proyecto existente de IntelliJ IDEA, agregar dependencias de Project Object Model (POM) y referenciar los plugins correspondientes.

Dependencias POM

<dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.4.2</version>
        </dependency>
        <!-- Apache Spark SQL -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.4.2</version>
        </dependency>
</dependencies>

Plugins

<build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.7.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <configuration>
                    <recompileMode>incremental</recompileMode>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

Código de ejemplo

package com.aliyun.emr.example.spark
import org.apache.spark.sql.SparkSession
object SparkMaxComputeDemo {
  def main(args: Array[String]): Unit = {
    // Create a SparkSession.
    val spark = SparkSession.builder()
      .appName("HelloDataWorks")
      .getOrCreate()
    // Print the Spark version.
    println(s"Spark version: ${spark.version}")
  }
}

Después de editar el código Scala, genere un paquete jar.

El paquete jar de ejemplo se llama SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar.

Cargue el recurso JAR.
1. Una vez finalizado el desarrollo del código, inicie sesión en la consola de OSS. En el panel de navegación izquierdo de la región, haga clic en Bucket list.
2. Haga clic en el nombre del bucket de destino para abrir la página File Management.
  
  Este tema utiliza el bucket onaliyun-bucket-2 como ejemplo.
3. Haga clic en Create Directory para crear un directorio donde almacenar el recurso JAR.
  
  Establezca Catalog Name como emr/jars para crear el directorio de almacenamiento del recurso JAR.
4. Cargue el recurso JAR en el directorio de almacenamiento.
  
  Vaya al directorio y haga clic en Upload File. En el área Files to Upload, haga clic en Scan Files, agregue el archivo SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar al bucket y luego haga clic en Upload File.

Referencie el recurso JAR.

Edite el código para referenciar el recurso JAR.

En la página de edición del nodo EMR Spark creado, modifique el código para referenciar el recurso JAR.

spark-submit --class com.aliyun.emr.example.spark.SparkMaxComputeDemo --master yarn ossref://onaliyun-bucket-2/emr/jars/SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar

Descripción de parámetros:

Parámetro	Descripción
`class`	El nombre completo de la clase principal que se va a ejecutar.
`master`	El modo de ejecución de la aplicación Spark.
Ruta de archivo `ossref`	El formato es `ossref://{endpoint}/{bucket}/{object}`. Endpoint: El endpoint del servicio OSS. Si este parámetro se deja vacío, el recurso de OSS debe estar en la misma región que el clúster EMR. Bucket: Un contenedor en OSS para almacenar objetos. Cada bucket tiene un nombre único. Inicie sesión en la consola de OSS para ver todos los buckets bajo su cuenta. Object: Un objeto específico, como un nombre de archivo o una ruta, almacenado en un bucket.

Ejecute el trabajo del nodo EMR Spark.

Una vez finalizada la edición, haga clic en el icono y seleccione el resource group Serverless que creó para ejecutar el nodo EMR Spark. Cuando el trabajo finalice, registre los applicationIds desde la consola; por ejemplo, application_1730367929285_xxxx.

Consulte el resultado.

Cree un nodo EMR Shell y ejecute el comando yarn logs -applicationId application_1730367929285_xxxx en el nodo para ver el resultado de la ejecución:

yarn logs -applicationId application_xxx
Process Output>>> LogLastModifiedTime:Fri Nov 01 16:27:46 +0800 2024
Process Output>>> LogLength:21
Process Output>>> LogContents:
Process Output>>> Spark version: 3.4.2
Process Output>>>

(Opcional) Configurar parámetros avanzados

Puede configurar los siguientes parámetros de propiedad única en el panel Scheduling Settings ubicado en el lado derecho del nodo, bajo EMR Node Parameters > DataWorks parameters.

Nota

Los parámetros avanzados disponibles varían según el tipo de clúster EMR, como se muestra en las siguientes tablas.
Para más propiedades de Spark de código abierto, puede configurarlas en el panel Scheduling Settings bajo EMR Node Parameters > Spark parameter.

Clúster Datalake/clúster personalizado: EMR on ECS

Parámetro avanzado	Descripción
queue	La cola de programación para el trabajo. El valor predeterminado es `default`. Si configuró una YARN Resource Queue a nivel de espacio de trabajo al registrar el clúster EMR en el espacio de trabajo de DataWorks: Si selecciona la casilla Global Settings Take Precedence y la establece en Yes, el trabajo de Spark utiliza la cola configurada al registrar el clúster EMR. Si no selecciona la casilla, el trabajo de Spark utiliza la cola configurada para el nodo EMR Spark. Para obtener más información sobre EMR YARN, consulte Basic queue configuration. Para obtener más información sobre la configuración de la cola al registrar un clúster EMR, consulte Set the global YARN resource queue.
priority	La prioridad. El valor predeterminado es 1.
FLOW_SKIP_SQL_ANALYZE	El modo de ejecución para instrucciones SQL. Valores válidos: `true`: Ejecuta varias instrucciones SQL a la vez. `false` (predeterminado): Ejecuta una instrucción SQL a la vez. Nota Este parámetro solo se admite para ejecuciones de prueba en el entorno de desarrollo de datos.
Otros	Puede agregar parámetros personalizados de Spark en Advanced Configurations. Por ejemplo, si agrega `spark.eventLog.enabled : false` , DataWorks añade automáticamente el parámetro al código que se envía al clúster EMR en formato `--conf key=value`. También puede configurar parámetros globales de Spark. Para obtener más información, consulte Set global Spark parameters. Nota Para habilitar el control de acceso Ranger, agregue la configuración `spark.hadoop.fs.oss.authorization.method=ranger` en Set global Spark parameters para asegurar que el control de acceso Ranger surta efecto.

Clúster Spark: EMR on ACK

Parámetro avanzado

Descripción

FLOW_SKIP_SQL_ANALYZE

El modo de ejecución para instrucciones SQL. Valores válidos:

true: Ejecuta varias instrucciones SQL a la vez.
false: Ejecuta una instrucción SQL a la vez.

Nota

Este parámetro solo se admite para ejecuciones de prueba en el entorno de desarrollo de datos.

Otros

Puede agregar parámetros personalizados de Spark en Advanced Configurations. Por ejemplo, spark.eventLog.enabled: false. DataWorks añade automáticamente el parámetro al código que se envía al clúster EMR en formato --conf key=value.
También puede configurar parámetros globales de Spark. Para obtener más información, consulte Set global Spark parameters.

Clúster Hadoop: EMR on ECS

Parámetro avanzado	Descripción
queue	La cola de programación a la que se envían los trabajos. Valor predeterminado: default. Si configura una YARN resource queue a nivel de espacio de trabajo al registrar un clúster EMR en un espacio de trabajo de DataWorks, se aplican las siguientes reglas: Si selecciona la casilla Global Settings Take Precedence y establece el valor en Yes, la configuración de la cola que especifique al registrar el clúster EMR tendrá prioridad cuando se ejecute un trabajo de Spark. Si desmarca la casilla, la configuración de la cola que especifique para el nodo EMR Spark tendrá prioridad cuando se ejecute un trabajo de Spark. Para obtener más información sobre EMR YARN, consulte Basic queue configurations. Para obtener más información sobre cómo configurar una cola al registrar un clúster EMR, consulte Set the global YARN resource queue.
priority	La prioridad. El valor predeterminado es 1.
FLOW_SKIP_SQL_ANALYZE	El modo de ejecución para instrucciones SQL. Valores válidos: `true`: Ejecuta varias instrucciones SQL a la vez. `false`: Ejecuta una instrucción SQL a la vez. Nota Este parámetro solo se admite para ejecuciones de prueba en el entorno de desarrollo de datos.
USE_GATEWAY	Especifica si se debe enviar el trabajo de este nodo a través de un clúster Gateway. Valores válidos: `true`: Envía el trabajo a través del clúster Gateway. `false`: No envía el trabajo a través de un clúster Gateway. De forma predeterminada, el trabajo se envía al nodo principal. Nota Si establece este parámetro en `true` pero el clúster del nodo no está asociado con un clúster Gateway, los envíos posteriores de trabajos EMR fallarán.
Otros	Puede agregar parámetros personalizados de Spark directamente en Advanced Configurations. Para una entrada como `spark.eventLog.enabled : false` , DataWorks la añade automáticamente al código enviado al clúster EMR en formato `--conf key=value`. También puede configurar parámetros globales de Spark. Para obtener más información, consulte Set global Spark parameters. Nota Para habilitar el control de acceso Ranger, agregue la configuración `spark.hadoop.fs.oss.authorization.method=ranger` en Set global Spark parameters para asegurar que el control de acceso Ranger surta efecto.

Clúster EMR Serverless Spark

Para obtener más información sobre la configuración de parámetros, consulte Set parameters for submitting a Spark job.

Parámetro avanzado	Descripción
queue	La cola de programación para el trabajo. El valor predeterminado es `dev_queue`.
priority	La prioridad. El valor predeterminado es 1.
FLOW_SKIP_SQL_ANALYZE	El modo de ejecución para instrucciones SQL. Valores válidos: `true`: Ejecuta varias instrucciones SQL a la vez. `false`: Ejecuta una instrucción SQL a la vez. Nota Este parámetro solo se admite para ejecuciones de prueba en el entorno de desarrollo de datos.
SERVERLESS_RELEASE_VERSION	La versión del motor Spark. De forma predeterminada, el trabajo utiliza la Default Engine Version configurada para el clúster en Management Center bajo Clusters. Puede configurar este parámetro para establecer diferentes versiones del motor para distintos trabajos.
SERVERLESS_QUEUE_NAME	La cola de recursos. De forma predeterminada, el trabajo utiliza la Default Resource Queue configurada para el clúster en Management Center bajo Clusters. Si tiene requisitos de aislamiento y gestión de recursos, puede agregar colas. Para obtener más información, consulte Manage resource queues.
Otros	Puede agregar parámetros personalizados de Spark en la configuración avanzada. Por ejemplo, si agrega `spark.eventLog.enabled : false` , DataWorks añade automáticamente el parámetro al código que se envía al clúster EMR en formato `--conf key=value`. También puede configurar parámetros globales de Spark. Para obtener más información, consulte Set global Spark parameters.

Ejecutar el trabajo de Spark

En el panel Run Configuration, vaya a la sección Compute Resource y configure los campos Compute Resource y DataWorks Resource Group.
Nota
- También puede configurar las CUs for Scheduling según los recursos que requiera la tarea. La CU predeterminada es 0.25.
- Para acceder a un origen de datos a través de Internet público o en una VPC, debe utilizar un resource group de programación con conectividad de red verificada hacia dicho origen. Para obtener más información, consulte Network connectivity solutions.
En el cuadro de diálogo de parámetros de la barra de herramientas, seleccione el origen de datos correspondiente y luego haga clic en Run.

Para ejecutar el trabajo del nodo periódicamente, configure sus propiedades de programación. Para obtener más información, consulte Configure scheduling properties for a node.

Nota
Si necesita personalizar el entorno del componente, puede crear una custom image basada en la imagen oficial dataworks_emr_base_task_pod y use the image in Data Development.

Por ejemplo, al crear una imagen personalizada, puede reemplazar paquetes JAR de Spark o depender de libraries, files o JAR packages específicos.
Después de configurar el trabajo del nodo, debe desplegarlo. Para obtener más información, consulte Deploy a node or workflow.
Una vez desplegado el trabajo, puede ir a Operation Center para consultar el estado de ejecución del trabajo periódico. Para obtener más información, consulte Get started with Operation Center.

Preguntas frecuentes

Why does the "DlfMetaStoreClientFactory not found" error occur when I run spark-submit in YARN-Cluster mode on an EMR Spark node after Kerberos is enabled in the EMR cluster?
P: ¿Cómo resuelvo un error de tiempo de espera en la conexión del nodo?

R: Asegúrese de que el resource group y el clúster tengan conectividad de red entre sí. Vaya a la lista de recursos de cómputo y haga clic en Resource Initialization. En el cuadro de diálogo que aparece, haga clic en Re-initialize. Compruebe que la inicialización se haya completado correctamente.

DataWorks:Nodo EMR Spark

Requisitos previos

Limitaciones

Notes

Desarrollar un trabajo de Spark y obtener el paquete JAR

Procedimiento

Desarrollar un trabajo de Spark

Opción 1: Cargar y hacer referencia

JAR < 500 MB

JAR >= 500 MB

Opción 2: Referenciar un recurso de OSS

Dependencias POM

Plugins

(Opcional) Configurar parámetros avanzados

Clúster Datalake/clúster personalizado: EMR on ECS

Clúster Spark: EMR on ACK

Clúster Hadoop: EMR on ECS

Clúster EMR Serverless Spark

Ejecutar el trabajo de Spark

Preguntas frecuentes

Temas relacionados