Exemplos do Spark 2.x - : Central de Documentação do Alibaba Cloud

Este tópico descreve como configurar as dependências do Spark 2.x e fornece alguns exemplos.

Configurar dependências para o Spark 2.x

Para enviar sua aplicação Spark 2.x usando o Spark on MaxCompute, adicione as seguintes dependências ao arquivo pom.xml. Para obter mais informações sobre o pom.xml, consulte pom.xml.

<properties>
    <spark.version>2.3.0</spark.version>
    <cupid.sdk.version>3.3.8-public</cupid.sdk.version>
    <scala.version>2.11.8</scala.version>
    <scala.binary.version>2.11</scala.binary.version>
</properties>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_${scala.binary.version}</artifactId>
    <version>${spark.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_${scala.binary.version}</artifactId>
    <version>${spark.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-mllib_${scala.binary.version}</artifactId>
    <version>${spark.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_${scala.binary.version}</artifactId>
    <version>${spark.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>cupid-sdk</artifactId>
    <version>${cupid.sdk.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>hadoop-fs-oss</artifactId>
    <version>${cupid.sdk.version}</version>
</dependency>
<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>odps-spark-datasource_${scala.binary.version}</artifactId>
    <version>${cupid.sdk.version}</version>
</dependency>
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>${scala.version}</version>
</dependency>
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-actors</artifactId>
    <version>${scala.version}</version>
</dependency>

No código anterior, defina o parâmetro scope conforme as instruções abaixo:

Defina scope como provided para todos os pacotes lançados pela comunidade Apache Spark, como spark-core e spark-sql.
Defina scope como compile para o módulo odps-spark-datasource.

Exemplo de WordCount (Scala)

Código de exemplo

WordCount.scala

Como enviar

cd /path/to/MaxCompute-Spark/spark-2.x
mvn clean package

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class \
    com.aliyun.odps.spark.examples.WordCount \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de leitura ou gravação de dados em uma tabela do MaxCompute (Scala)

Código de exemplo

SparkSQL.scala

Como enviar

cd /path/to/MaxCompute-Spark/spark-2.x
mvn clean package
# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.sparksql.SparkSQL \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de GraphX PageRank (Scala)

Código de exemplo

PageRank.scala

Como enviar

cd /path/to/MaxCompute-Spark/spark-2.x
mvn clean package
# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.graphx.PageRank \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de MLlib KMeans-ON-OSS (Scala)

Para obter informações sobre como configurar spark.hadoop.fs.oss.ststoken.roleArn e spark.hadoop.fs.oss.endpoint, consulte Notas de acesso ao OSS.

Código de exemplo

KmeansModelSaveToOss.scala

Como enviar

# Edit the code. 
val modelOssDir = "oss://bucket/kmeans-model" // Enter the path of the OSS bucket. 
val spark = SparkSession
  .builder()
  .config("spark.hadoop.fs.oss.credentials.provider", "org.apache.hadoop.fs.aliyun.oss.AliyunStsTokenCredentialsProvider")
  .config("spark.hadoop.fs.oss.ststoken.roleArn", "acs:ram::****:role/aliyunodpsdefaultrole")
  .config("spark.hadoop.fs.oss.endpoint", "oss-cn-hangzhou-zmf.aliyuncs.com")
  .appName("KmeansModelSaveToOss")
  .getOrCreate()

cd /path/to/MaxCompute-Spark/spark-2.x
mvn clean package
# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.mllib.KmeansModelSaveToOss \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de OSS UnstructuredData (Scala)

Para obter informações sobre como configurar spark.hadoop.fs.oss.ststoken.roleArn e spark.hadoop.fs.oss.endpoint, consulte Notas de acesso ao OSS.

Código de exemplo

SparkUnstructuredDataCompute.scala

Como enviar

# Edit the code. 
val pathIn = "oss://bucket/inputdata/" // Enter the path of the OSS bucket. 
val spark = SparkSession
  .builder()
  .config("spark.hadoop.fs.oss.credentials.provider", "org.apache.hadoop.fs.aliyun.oss.AliyunStsTokenCredentialsProvider")
  .config("spark.hadoop.fs.oss.ststoken.roleArn", "acs:ram::****:role/aliyunodpsdefaultrole")
  .config("spark.hadoop.fs.oss.endpoint", "oss-cn-hangzhou-zmf.aliyuncs.com")
  .appName("SparkUnstructuredDataCompute")
  .getOrCreate()

cd /path/to/MaxCompute-Spark/spark-2.x
mvn clean package
# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.oss.SparkUnstructuredDataCompute \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo SparkPi (Scala)

Código de exemplo

SparkPi.scala

Como enviar

cd /path/to/MaxCompute-Spark/spark-2.x
mvn clean package

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.SparkPi \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de Spark Streaming LogHub (Scala)

Código de exemplo

LogHubStreamingDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.streaming.loghub.LogHubStreamingDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de uso do Spark Streaming LogHub para gravar dados no MaxCompute (Scala)

Código de exemplo

LogHub2OdpsDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.streaming.loghub.LogHub2OdpsDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de Spark Streaming DataHub (Scala)

Código de exemplo

DataHubStreamingDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.streaming.datahub.DataHubStreamingDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de uso do Spark Streaming DataHub para gravar dados no MaxCompute (Scala)

Código de exemplo

DataHub2OdpsDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.streaming.datahub.DataHub2OdpsDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de Spark Streaming Kafka (Scala)

Código de exemplo

KafkaStreamingDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.streaming.kafka.KafkaStreamingDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Nota

Para obter mais informações, consulte Spark on MaxCompute.

Exemplo de Spark StructuredStreaming DataHub (Scala)

Código de exemplo

DatahubStructuredStreamingDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.structuredstreaming.datahub.DatahubStructuredStreamingDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de Spark StructuredStreaming Kafka (Scala)

Código de exemplo

KafkaStructuredStreamingDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.structuredstreaming.kafka.KafkaStructuredStreamingDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de Spark StructuredStreaming LogHub (Scala)

Código de exemplo

LoghubStructuredStreamingDemo.scala

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.structuredstreaming.loghub.LoghubStructuredStreamingDemo \
    /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

Exemplo de uso do PySpark para ler ou gravar dados em uma tabela do MaxCompute (Python)

Código de exemplo

spark_sql.py

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --jars /path/to/odps-spark-datasource_2.11-3.3.8-public.jar \
    /path/to/MaxCompute-Spark/spark-2.x/src/main/python/spark_sql.py

Exemplo de uso do PySpark para gravar dados no OSS (Python)

Código de exemplo

spark_oss.py

Como enviar

# For information about how to configure the environment variables in the spark-defaults.conf file, see Set up a Spark on MaxCompute development environment. 
# For information about OSS configurations, see OSS access notes. 

cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --jars /path/to/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar \
    /path/to/MaxCompute-Spark/spark-2.x/src/main/python/spark_oss.py
# Compile Spark 2.x to obtain the spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar package.

Exemplo de Spark SQL (Java)

Para obter mais informações sobre o código de exemplo Java do Spark SQL, consulte JavaSparkSQL.java.

Exemplo de leitura de dados do MaxCompute e gravação no HBase

Use o IntelliJ IDEA para escrever código que leia dados do MaxCompute e os grave no HBase.

Código de exemplo

object McToHbase {
  def main(args: Array[String]) {
    val spark = SparkSession
      .builder()
      .appName("spark_sql_ddl")
      .config("spark.sql.catalogImplementation", "odps")
      .config("spark.hadoop.odps.end.point","http://service.cn.maxcompute.aliyun.com/api")
      .config("spark.hadoop.odps.runtime.end.point","http://service.cn.maxcompute.aliyun-inc.com/api")
      .getOrCreate()
      val sc = spark.sparkContext
      val config = HBaseConfiguration.create()
      val zkAddress = ""
      config.set(HConstants.ZOOKEEPER_QUORUM, zkAddress);
      val jobConf = new JobConf(config)
      jobConf.setOutputFormat(classOf[TableOutputFormat])
      jobConf.set(TableOutputFormat.OUTPUT_TABLE,"test")

    try{
      import spark._
      spark.sql("select '7', 'long'").rdd.map(row => {
        val id = row(0).asInstanceOf[String]
        val name = row(1).asInstanceOf[String]
        val put = new Put(Bytes.toBytes(id))
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("a"), Bytes.toBytes(name))
        (new ImmutableBytesWritable, put)
    }).saveAsHadoopDataset(jobConf)
  } finally {
    sc.stop()
  }

  }
}

Como enviar: Use o IntelliJ IDEA para enviar e executar o código de exemplo. Para obter mais informações, consulte Modos de execução do Spark on MaxCompute.

Exemplos de leitura e gravação de dados em objetos do OSS

Use o IntelliJ IDEA ou o DataWorks para ler dados de objetos do OSS e gravá-los.

Código de exemplo

Exemplo 1: Código de exemplo para o modo local

package com.aliyun.odps.spark.examples
import java.io.ByteArrayInputStream
import org.apache.spark.sql.SparkSession

object SparkOSS {
  def main(args: Array[String]) {
    val spark = SparkSession
      .builder()
      .config("spark.master", "local[4]") // The code can run after you set spark.master to local[N]. N indicates the number of concurrent Spark jobs. 
      .config("spark.hadoop.fs.oss.accessKeyId", "")
      .config("spark.hadoop.fs.oss.accessKeySecret", "")
      .config("spark.hadoop.fs.oss.endpoint", "oss-cn-beijing.aliyuncs.com")
      .appName("SparkOSS")
      .getOrCreate()

    val sc = spark.sparkContext
    try {
      // Read data from OSS objects. 
      val pathIn = "oss://spark-oss/workline.txt"
      val inputData = sc.textFile(pathIn, 5)
            // Write Resilient Distributed Datasets (RDD). 
      inputData.repartition(1).saveAsTextFile("oss://spark-oss/user/data3")

    } finally {
      sc.stop()
    }
  }
}

Nota

Antes de executar o código de exemplo, verifique se a dependência hadoop-fs-oss foi adicionada. Caso contrário, um erro será retornado.

Exemplo 2: Código de exemplo para o modo local

package com.aliyun.odps.spark.examples
import java.io.ByteArrayInputStream
import com.aliyun.oss.{OSSClientBuilder,OSSClient}
import org.apache.spark.sql.SparkSession

object SparkOSS {
  def main(args: Array[String]) {
    val spark = SparkSession
      .builder()
      .config("spark.master", "local[4]") // The code can run after you set spark.master to local[N]. N indicates the number of concurrent Spark jobs. 
      .config("spark.hadoop.fs.oss.accessKeyId", "")
      .config("spark.hadoop.fs.oss.accessKeySecret", "")
      .config("spark.hadoop.fs.oss.endpoint", "oss-cn-beijing.aliyuncs.com")
      .appName("SparkOSS")
      .getOrCreate()

    val sc = spark.sparkContext
    try {
      // Read data from OSS objects. 
      val pathIn = "oss://spark-oss/workline.txt"
      val inputData = sc.textFile(pathIn, 5)
      val cnt = inputData.count
      inputData.count()
      println(s"count: $cnt")

      // Write data to OSS objects. 
      // The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. Using these credentials to perform operations is a high-risk operation. We recommend that you use a RAM user to call API operations or perform routine O&M. To create a RAM user, log on to the RAM console.
		  // In this example, the AccessKey ID and AccessKey secret are configured as environment variables. You can also save your AccessKey pair in the configuration file based on your business requirements.
			// We recommend that you do not directly specify the AccessKey ID and AccessKey secret in code to prevent AccessKey pair leaks.
      val ossClient = new OSSClientBuilder().build("oss-cn-beijing.aliyuncs.com", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
      val filePath="user/data"
      ossClient.putObject("spark-oss",filePath , new ByteArrayInputStream(cnt.toString.getBytes()))
      ossClient.shutdown()
    } finally {
      sc.stop()
    }
  }
}

Exemplo 3: Código de exemplo para o modo cluster

package com.aliyun.odps.spark.examples
import java.io.ByteArrayInputStream
import com.aliyun.oss.{OSSClientBuilder,OSSClient}
import org.apache.spark.sql.SparkSession

object SparkOSS {
  def main(args: Array[String]) {
    val spark = SparkSession
      .builder()
      .appName("SparkOSS")
      .getOrCreate()

    val sc = spark.sparkContext
    try {
      // Read data from OSS objects. 
      val pathIn = "oss://spark-oss/workline.txt"
      val inputData = sc.textFile(pathIn, 5)
      val cnt = inputData.count
      inputData.count()
      println(s"count: $cnt")

      // inputData.repartition(1).saveAsTextFile("oss://spark-oss/user/data3")
      // Write data to OSS objects. 
      // The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. Using these credentials to perform operations is a high-risk operation. We recommend that you use a RAM user to call API operations or perform routine O&M. To create a RAM user, log on to the RAM console.
			// In this example, the AccessKey ID and AccessKey secret are configured as environment variables. You can also save your AccessKey pair in the configuration file based on your business requirements.
			// We recommend that you do not directly specify the AccessKey ID and AccessKey secret in code to prevent AccessKey pair leaks.
      val ossClient = new OSSClientBuilder().build("oss-cn-beijing.aliyuncs.com", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
      val filePath="user/data"
      ossClient.putObject("spark-oss",filePath , new ByteArrayInputStream(cnt.toString.getBytes()))
      ossClient.shutdown()
    } finally {
      sc.stop()
    }
  }
}

Como enviar
- Use o IntelliJ IDEA para desenvolver, testar e enviar o código para o modo local. Para obter mais informações, consulte Modos de execução do Spark on MaxCompute.
- Use um nó ODPS Spark para enviar e executar o código no console do DataWorks. Para obter mais informações, consulte Desenvolver uma tarefa ODPS Spark.

Exemplo de leitura de dados do MaxCompute e gravação no OSS

Use o IntelliJ IDEA ou o DataWorks para ler dados do MaxCompute e gravá-los no OSS.

Código de exemplo

Código de exemplo para o modo local

package com.aliyun.odps.spark.examples.userpakage

import org.apache.spark.sql.{SaveMode, SparkSession}

object SparkODPS2OSS {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("Spark2OSS")
      .config("spark.master", "local[4]")// The code can run after you set spark.master to local[N]. N indicates the number of concurrent Spark jobs.
      .config("spark.hadoop.odps.project.name", "")
      .config("spark.hadoop.odps.access.id", "")
      .config("spark.hadoop.odps.access.key", "")
      .config("spark.hadoop.odps.end.point", "http://service.cn.maxcompute.aliyun.com/api")
      .config("spark.sql.catalogImplementation", "odps")
      .config("spark.hadoop.fs.oss.accessKeyId","")
      .config("spark.hadoop.fs.oss.accessKeySecret","")
      .config("spark.hadoop.fs.oss.endpoint","oss-cn-beijing.aliyuncs.com")
      .getOrCreate()

    try{

      // Use Spark SQL to query tables.
      val data = spark.sql("select * from  user_detail")
     // Show the query results.
      data.show(10)
      // Store the query results to an OSS object.
      data.toDF().coalesce(1).write.mode(SaveMode.Overwrite).csv("oss://spark-oss/user/data3")
    }finally {
      spark.stop()
    }

  }
}

Código de exemplo para o modo cluster

package com.aliyun.odps.spark.examples.userpakage
import org.apache.spark.sql.{SaveMode, SparkSession}

object SparkODPS2OSS {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("SparkODPS2OSS")
      .getOrCreate()

    try{

      // Use Spark SQL to query tables.
      val data = spark.sql("select * from  user_detail")
     // Show the query results.
      data.show(10)
      // Store the query results to an OSS object.
      data.toDF().coalesce(1).write.mode(SaveMode.Overwrite).csv("oss://spark-oss/user/data3")

    }finally {
      spark.stop()
    }

  }
}

Como enviar
- Use o IntelliJ IDEA para desenvolver, testar e enviar o código para o modo local.
- Use um nó ODPS Spark para enviar e executar o código no console do DataWorks. Para obter mais informações, consulte Desenvolver uma tarefa ODPS Spark.
Nota
Para obter mais informações sobre como configurar o ambiente de desenvolvimento do Spark, consulte Modos de execução do Spark on MaxCompute.