Uso de Spark para acceder a MaxCompute - E-MapReduce - Centro de documentación de Alibaba Cloud

Apache Spark en E-MapReduce (EMR) permite leer y escribir datos en MaxCompute mediante la clase OdpsOps.

Requisitos previos

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

Un clúster de EMR con Spark habilitado
Un proyecto y una tabla de MaxCompute
Un AccessKey ID y un AccessKey Secret de Alibaba Cloud

Importante

Almacene sus credenciales como variables de entorno en lugar de incluirlas directamente en el código. Las credenciales integradas en el código pueden exponerse accidentalmente en sistemas de control de versiones o registros.

Funcionamiento

La clase OdpsOps forma parte del SDK de EMR y gestiona datos de MaxCompute desde Spark. Para transferir datos entre Spark y MaxCompute:

Inicialice un objeto OdpsOps con sus credenciales y los endpoints de MaxCompute.
Llame a readTable para cargar una tabla de MaxCompute en un RDD (Resilient Distributed Dataset) de Spark.
Procese los datos en Spark.
Llame a saveToTable para escribir el RDD resultante en una tabla de MaxCompute.

Lectura y escritura de datos en MaxCompute

Paso 1: Inicializar OdpsOps

Establezca sus credenciales como variables de entorno; luego, importe las clases necesarias y cree un objeto OdpsOps:

export ALIBABA_CLOUD_ACCESS_KEY_ID=<your-access-key-id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<your-access-key-secret>

import com.aliyun.odps.TableSchema
import com.aliyun.odps.data.Record
import org.apache.spark.aliyun.odps.OdpsOps
import org.apache.spark.{SparkContext, SparkConf}

object Sample {
  def main(args: Array[String]): Unit = {

    // Read credentials from environment variables
    val accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    val accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")

    // Internal network endpoints for MaxCompute
    val odpsEndpoint = "http://odps-ext.aliyun-inc.com/api"
    val tunnelEndpoint = "http://dt-ext.odps.aliyun-inc.com"

    val conf = new SparkConf().setAppName("Test Odps")
    val sc = new SparkContext(conf)
    val odpsOps = OdpsOps(sc, accessKeyId, accessKeySecret, odpsEndpoint, tunnelEndpoint)

    // Step 2: read data
    // Step 3: write data
  }
}

Parámetros del constructor de OdpsOps

Parámetro	Descripción
`sc`	La instancia de `SparkContext`
`accessKeyId`	Su AccessKey ID de Alibaba Cloud, obtenido de una variable de entorno
`accessKeySecret`	Su AccessKey Secret de Alibaba Cloud, obtenido de una variable de entorno
`odpsEndpoint`	El endpoint del servicio MaxCompute
`tunnelEndpoint`	El endpoint de MaxCompute Tunnel para la transferencia de datos

Paso 2: Cargar datos desde MaxCompute

Use readTable para cargar una tabla de MaxCompute en un RDD de Spark. Defina una función read que asigne cada Record al tipo de Scala que necesite:

// Define how to parse each MaxCompute record into a Scala value
def read(record: Record, schema: TableSchema): String = {
  record.getString(0)
}

val project = "<odps-project>"
val table = "<odps-table>"
val numPartitions = 2

val inputData = odpsOps.readTable(project, table, read, numPartitions)
inputData.top(10).foreach(println)

Parámetros de readTable

Parámetro	Descripción
`project`	El nombre del proyecto de MaxCompute
`table`	El nombre de la tabla de MaxCompute
`read`	Una función con la firma `(Record, TableSchema) => T` que analiza sintácticamente cada registro
`numPartitions`	El número de particiones del RDD para lecturas paralelas

Paso 3: Escribir datos en MaxCompute

Use saveToTable para escribir un RDD de Spark en una tabla de MaxCompute. Defina una función write que llene un Record con cada elemento del RDD:

// Define how to convert each Scala value into a MaxCompute record
def write(s: String, emptyRecord: Record, schema: TableSchema): Unit = {
  val r = emptyRecord
  r.set(0, s)
}

val resultData = inputData.map(e => s"$e has been processed.")
odpsOps.saveToTable(project, table, resultData, write)

Parámetros de saveToTable

Parámetro	Descripción
`project`	El nombre del proyecto de MaxCompute
`table`	El nombre de la tabla de MaxCompute
`dataRDD`	El RDD de Spark que se va a escribir
`write`	Una función con la firma `(T, Record, TableSchema) => Unit` que llena cada registro

Trabajo con tablas particionadas

Al leer o escribir en una tabla particionada, pase la especificación de partición como <partition-key>=<partition-value>. Separe varias particiones con comas.

Especificación de partición	Ejemplo
Partición única	`pt='1'`
Varias particiones	`pt='1', ps='2'`

Siguientes pasos

Para ver el código de ejemplo completo, consulte SparkMaxComputeDemo en GitHub.