How to use the Spark RDD API to develop offline jobs - E-MapReduce

Utilice Spark para leer datos de Simple Log Service (SLS) en modo por lotes (offline), como un conjunto de datos acotado con hora de inicio y fin definidas, en lugar de hacerlo como un flujo continuo. EMR admite dos enfoques: Spark RDD y Spark SQL.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Un clúster de EMR con Spark instalado
Un proyecto de Simple Log Service y un logstore que contengan los datos a leer
Un AccessKey ID y un AccessKey secret de un usuario de RAM con permisos de lectura en el logstore
(Solo para Spark SQL) Acceso al nodo del clúster de EMR donde se encuentran los archivos JAR

Uso de Spark RDD para leer desde Simple Log Service

Código de ejemplo

El siguiente programa en Scala lee datos de registros de un logstore para un intervalo de tiempo especificado y guarda la salida en una ruta del sistema de archivos.

// TestBatchLoghub.scala

object TestBatchLoghub {
  def main(args: Array[String]): Unit = {
    if (args.length < 6) {
      System.err.println(
        """Usage: TestBatchLoghub <sls project> <sls logstore> <sls endpoint>
          |  <access key id> <access key secret> <output path> <start time> <end time=now>
        """.stripMargin)
      System.exit(1)
    }

    val loghubProject    = args(0)
    val logStore         = args(1)
    val endpoint         = args(2)
    val accessKeyId      = args(3)     // Read from environment variable at submit time
    val accessKeySecret  = args(4)     // Read from environment variable at submit time
    val outputPath       = args(5)
    val startTime        = args(6).toLong

    val sc = new SparkContext(new SparkConf().setAppName("test batch loghub"))
    var rdd: JavaRDD[String] = null

    if (args.length > 7) {
      // Read log data between startTime and endTime
      rdd = LoghubUtils.createRDD(sc, loghubProject, logStore, accessKeyId, accessKeySecret, endpoint, startTime, args(7).toLong)
    } else {
      // Read log data from startTime to now
      rdd = LoghubUtils.createRDD(sc, loghubProject, logStore, accessKeyId, accessKeySecret, endpoint, startTime)
    }

    rdd.saveAsTextFile(outputPath)
  }
}

LoghubUtils.createRDD() devuelve un objeto JavaRDD[String] donde cada elemento es una entrada de registro. Pase siete argumentos para leer hasta la hora actual, u ocho argumentos para especificar una hora de fin explícita.

Para consultar la configuración del POM de Maven, consulte aliyun-emapreduce-demo.

Parámetros de conexión

Parámetro	Descripción	Ejemplo
`<sls project>`	Nombre del proyecto de SLS	`my-project`
`<sls logstore>`	Nombre del logstore dentro del proyecto	`my-logstore`
`<sls endpoint>`	Endpoint de SLS para su región	`cn-hangzhou.log.aliyuncs.com`
`<access key id>`	AccessKey ID de su usuario de RAM	Utilice `$ALIBABA_CLOUD_ACCESS_KEY_ID`
`<access key secret>`	AccessKey secret de su usuario de RAM	Utilice `$ALIBABA_CLOUD_ACCESS_KEY_SECRET`
`<output path>`	Ruta de salida para el resultado del RDD	`oss://my-bucket/output/`
`<start time>`	Inicio del intervalo de tiempo	`1700000000`
`<end time>`	(Opcional) Fin del intervalo de tiempo. El valor predeterminado es la hora actual.	`1700003600`

Compilación y ejecución

Paso 1: Compile el código.

mvn clean package -DskipTests

El archivo JAR compilado se guarda en el directorio target/shaded/.

Paso 2: Envíe el trabajo.

Importante

Utilice el par de AccessKey de un usuario de RAM en lugar del par de AccessKey de su cuenta de Alibaba Cloud. Un par de AccessKey a nivel de cuenta otorga acceso a todas las operaciones de la API. Para obtener información sobre cómo crear un usuario de RAM, consulte Create a RAM user. Almacene las credenciales como variables de entorno; no las codifique directamente en sus scripts.

Establezca las variables de entorno antes de enviar el trabajo:

export ALIBABA_CLOUD_ACCESS_KEY_ID=<access_key_id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<access_key_secret>

A continuación, envíe el trabajo:

spark-submit \
  --master yarn-cluster \
  --executor-cores 2 \
  --executor-memory 1g \
  --driver-memory 1g \
  --num-executors 2 \
  --class x.x.x.TestBatchLoghub xxx.jar \
  <sls project> <sls logstore> <sls endpoint> \
  $ALIBABA_CLOUD_ACCESS_KEY_ID $ALIBABA_CLOUD_ACCESS_KEY_SECRET \
  <output path> <start time> [<end time>]

Reemplace x.x.x.TestBatchLoghub con el nombre completo real de la clase y xxx.jar con la ruta real del archivo JAR. Ajuste los valores de --executor-cores, --executor-memory y --num-executors según el volumen de datos y la capacidad de su clúster.

Uso de Spark SQL para leer desde Simple Log Service

Spark SQL utiliza el origen de datos loghub, que se incluye en el JAR de extensión de Spark de EMR.

Inicio de spark-sql con el JAR de LogHub

Importante

Utilice el par de AccessKey de un usuario de RAM en lugar del par de AccessKey de su cuenta de Alibaba Cloud. Almacene las credenciales como variables de entorno pasadas mediante --hiveconf; no las codifique directamente en sus scripts.

spark-sql \
  --jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/* \
  --hiveconf accessKeyId=$ALIBABA_CLOUD_ACCESS_KEY_ID \
  --hiveconf accessKeySecret=$ALIBABA_CLOUD_ACCESS_KEY_SECRET

Si su clúster de EMR utiliza Spark 2, reemplace spark3 por spark2 en la ruta del JAR:

/opt/apps/SPARK-EXTENSION/spark-extension-current/spark2-emrsdk/*

Creación de tablas y consulta de datos

CREATE TABLE test_sls
USING loghub
OPTIONS (
  endpoint         = 'cn-hangzhou-intranet.log.aliyuncs.com',
  access.key.id    = '${hiveconf:accessKeyId}',
  access.key.secret= '${hiveconf:accessKeySecret}',
  sls.project      = 'test_project',
  sls.store        = 'test_store',
  startingoffsets  = 'earliest'
);

SELECT * FROM test_sls;

Parámetros de conexión

Parámetro	Obligatorio	Descripción	Ejemplo
`endpoint`	Sí	Endpoint de SLS para su región	`cn-hangzhou-intranet.log.aliyuncs.com`
`access.key.id`	Sí	AccessKey ID, pasado mediante `--hiveconf`	`${hiveconf:accessKeyId}`
`access.key.secret`	Sí	AccessKey secret, pasado mediante `--hiveconf`	`${hiveconf:accessKeySecret}`
`sls.project`	Sí	Nombre del proyecto de SLS	`test_project`
`sls.store`	Sí	Nombre del logstore	`test_store`
`startingoffsets`	No	Posición inicial de lectura. Utilice `earliest` para leer todos los datos disponibles desde el inicio del logstore.	`earliest`

Uso del JAR de LogHub en un entorno de desarrollo local

Para desarrollar y probar localmente con Spark 3 (los pasos son los mismos para Spark 2), instale el JAR del origen de datos de EMR en su repositorio local de Maven.

Paso 1: Descargue el JAR desde su clúster de EMR.

Copie el JAR desde la siguiente ruta en su nodo de clúster a su computadora local:

/opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/emr-datasources_shaded_2.12

Paso 2: Instale el JAR en su repositorio local de Maven.

mvn install:install-file \
  -DgroupId=com.aliyun.emr \
  -DartifactId=emr-datasources_shaded_2.12 \
  -Dversion=3.0.2 \
  -Dpackaging=jar \
  -Dfile=<path-to-downloaded-jar>

Reemplace <path-to-downloaded-jar> con la ruta local donde guardó el JAR.

**Paso 3: Agregue la dependencia a su archivo pom.xml.**

<dependency>
  <groupId>com.aliyun.emr</groupId>
  <artifactId>emr-datasources_shaded_2.12</artifactId>
  <version>3.0.2</version>
</dependency>

Referencias

Para obtener más información sobre cómo utilizar Spark para acceder a Kafka, consulte la Structured Streaming + Kafka Integration Guide.