Use Spark to access MaxCompute - E-MapReduce - Centre de documentation Alibaba Cloud

Prérequis

Avant de commencer, vérifiez que vous disposez des éléments suivants :

Un cluster EMR avec Spark activé
Un projet et une table MaxCompute
Un ID AccessKey et un AccessKey Secret Alibaba Cloud

Important

Stockez vos identifiants sous forme de variables d'environnement plutôt que de les coder en dur. Les identifiants codés en dur risquent d'être exposés accidentellement dans le contrôle de version ou les journaux.

Fonctionnement

La classe OdpsOps du SDK EMR gère les données MaxCompute dans Spark. Pour transférer des données entre Spark et MaxCompute :

Initialisez un objet OdpsOps avec vos identifiants et les endpoints MaxCompute.
Appelez la méthode readTable pour charger une table MaxCompute dans un RDD Spark (Resilient Distributed Dataset).
Traitez les données dans Spark.
Appelez la méthode saveToTable pour écrire le RDD résultant dans une table MaxCompute.

Lire et écrire des données MaxCompute

Étape 1 : Initialiser OdpsOps

Définissez vos identifiants comme variables d'environnement, importez les classes requises et créez un objet 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
  }
}

Paramètres du constructeur OdpsOps

Paramètre	Description
`sc`	L'instance `SparkContext`
`accessKeyId`	Votre ID AccessKey Alibaba Cloud, lu depuis une variable d'environnement
`accessKeySecret`	Votre AccessKey Secret Alibaba Cloud, lu depuis une variable d'environnement
`odpsEndpoint`	L'endpoint du service MaxCompute
`tunnelEndpoint`	L'endpoint MaxCompute Tunnel pour le transfert de données

Étape 2 : Charger des données depuis MaxCompute

Utilisez la méthode readTable pour charger une table MaxCompute dans un RDD Spark. Définissez une fonction read qui convertit chaque Record vers le type Scala souhaité :

// 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)

Paramètres de readTable

Paramètre	Description
`project`	Le nom du projet MaxCompute
`table`	Le nom de la table MaxCompute
`read`	Une fonction de signature `(Record, TableSchema) => T` qui analyse chaque enregistrement
`numPartitions`	Le nombre de partitions RDD pour les lectures parallèles

Étape 3 : Écrire des données dans MaxCompute

Utilisez la méthode saveToTable pour écrire un RDD Spark dans une table MaxCompute. Définissez une fonction write qui remplit un Record à partir de chaque élément du 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)

Paramètres de saveToTable

Paramètre	Description
`project`	Le nom du projet MaxCompute
`table`	Le nom de la table MaxCompute
`dataRDD`	Le RDD Spark à écrire
`write`	Une fonction de signature `(T, Record, TableSchema) => Unit` qui remplit chaque enregistrement

Travailler avec des tables partitionnées

Lors de la lecture ou de l'écriture dans une table partitionnée, indiquez la spécification de partition sous la forme <partition-key>=<partition-value>. Séparez les partitions multiples par des virgules.

Spécification de partition	Exemple
Partition unique	`pt='1'`
Partitions multiples	`pt='1', ps='2'`

Étapes suivantes

Pour consulter l'exemple de code complet, rendez-vous sur la page SparkMaxComputeDemo sur GitHub.