Use Spark to access MaxCompute - E-MapReduce: Central de Documentação do Alibaba Cloud

Pré-requisitos

Antes de começar, verifique se você tem:

Um cluster EMR com o Spark ativado
Um projeto e uma tabela do MaxCompute
Um AccessKey ID e um AccessKey Secret da Alibaba Cloud

Importante

Armazene suas credenciais como variáveis de ambiente em vez de codificá-las diretamente no código. Credenciais codificadas podem ficar expostas acidentalmente em controles de versão ou logs.

Como funciona

A classe OdpsOps integra o SDK do EMR para gerenciar dados do MaxCompute no Spark. Para transferir dados entre o Spark e o MaxCompute:

Inicialize um objeto OdpsOps com suas credenciais e os endpoints do MaxCompute.
Chame readTable para carregar uma tabela do MaxCompute em um RDD (Resilient Distributed Dataset) do Spark.
Processe os dados no Spark.
Chame saveToTable para gravar o RDD resultante em uma tabela do MaxCompute.

Leitura e gravação de dados no MaxCompute

Etapa 1: Inicializar o OdpsOps

Defina suas credenciais como variáveis de ambiente, importe as classes necessárias e crie um 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 do construtor OdpsOps

Parâmetro	Descrição
`sc`	Instância de `SparkContext`
`accessKeyId`	AccessKey ID da Alibaba Cloud, lido de uma variável de ambiente
`accessKeySecret`	AccessKey Secret da Alibaba Cloud, lido de uma variável de ambiente
`odpsEndpoint`	Endpoint do serviço MaxCompute
`tunnelEndpoint`	Endpoint do MaxCompute Tunnel para transferência de dados

Etapa 2: Carregar dados do MaxCompute

Use readTable para carregar uma tabela do MaxCompute em um RDD do Spark. Defina uma função read que mapeie cada Record para o tipo Scala desejado:

// 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	Descrição
`project`	Nome do projeto no MaxCompute
`table`	Nome da tabela no MaxCompute
`read`	Função com assinatura `(Record, TableSchema) => T` que analisa cada registro
`numPartitions`	Número de partições do RDD para leituras paralelas

Etapa 3: Gravar dados no MaxCompute

Use saveToTable para gravar um RDD do Spark em uma tabela do MaxCompute. Defina uma função write que preencha um Record com base em cada elemento do 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	Descrição
`project`	Nome do projeto no MaxCompute
`table`	Nome da tabela no MaxCompute
`dataRDD`	RDD do Spark a ser gravado
`write`	Função com assinatura `(T, Record, TableSchema) => Unit` que preenche cada registro

Trabalho com tabelas particionadas

Ao ler ou gravar em uma tabela particionada, especifique a partição no formato <partition-key>=<partition-value>. Separe várias partições com vírgulas.

Especificação de partição	Exemplo
Partição única	`pt='1'`
Várias partições	`pt='1', ps='2'`

Próximas etapas

Para obter o código de exemplo completo, consulte SparkMaxComputeDemo no GitHub.