Spark 2.x örnekleri - - Alibaba Cloud Dokümantasyon Merkezi

Bu konu, Spark 2.x bağımlılıklarının nasıl yapılandırılacağını açıklar ve bazı örnekler sunar.

Spark 2.x için bağımlılıkları yapılandırın

Spark on MaxCompute kullanarak Spark 2.x uygulamanızı göndermek istiyorsanız, aşağıdaki bağımlılıkları pom.xml dosyasına eklemeniz gerekir. pom.xml hakkında daha fazla bilgi için bkz. 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>

Yukarıdaki kodda, scope parametresini aşağıdaki talimatlara göre ayarlayın:

spark-core ve spark-sql gibi Apache Spark topluluğunda yayınlanan tüm paketler için scope değerini provided olarak ayarlayın.
odps-spark-datasource modülü için scope değerini compile olarak ayarlayın.

WordCount örneği (Scala)

Örnek kod

WordCount.scala

Nasıl gönderilir

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

MaxCompute tablosundan veri okuma veya MaxCompute tablosuna veri yazma örneği (Scala)

Örnek kod

SparkSQL.scala

Nasıl gönderilir

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

GraphX PageRank örneği (Scala)

Örnek kod

PageRank.scala

Nasıl gönderilir

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

MLlib KMeans-ON-OSS örneği (Scala)

spark.hadoop.fs.oss.ststoken.roleArn ve spark.hadoop.fs.oss.endpoint öğelerinin nasıl yapılandırılacağı hakkında bilgi için bkz. OSS access notes.

Örnek kod

KmeansModelSaveToOss.scala

Nasıl gönderilir

# 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

OSS UnstructuredData örneği (Scala)

spark.hadoop.fs.oss.ststoken.roleArn ve spark.hadoop.fs.oss.endpoint öğelerinin nasıl yapılandırılacağı hakkında bilgi için bkz. OSS access notes.

Örnek kod

SparkUnstructuredDataCompute.scala

Nasıl gönderilir

# 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

SparkPi örneği (Scala)

Örnek kod

SparkPi.scala

Nasıl gönderilir

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

Spark Streaming LogHub örneği (Scala)

Örnek kod

LogHubStreamingDemo.scala

Nasıl gönderilir

# 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

Spark Streaming LogHub kullanarak verileri MaxCompute'a yazma örneği (Scala)

Örnek kod

LogHub2OdpsDemo.scala

Nasıl gönderilir

# 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

Spark Streaming DataHub örneği (Scala)

Örnek kod

DataHubStreamingDemo.scala

Nasıl gönderilir

# 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

Spark Streaming DataHub kullanarak verileri MaxCompute'a yazma örneği (Scala)

Örnek kod

DataHub2OdpsDemo.scala

Nasıl gönderilir

# 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

Spark Streaming Kafka örneği (Scala)

Örnek kod

KafkaStreamingDemo.scala

Nasıl gönderilir

# 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

Not

Daha fazla bilgi için bkz. Spark on MaxCompute.

Spark StructuredStreaming DataHub örneği (Scala)

Örnek kod

DatahubStructuredStreamingDemo.scala

Nasıl gönderilir

# 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

Spark StructuredStreaming Kafka örneği (Scala)

Örnek kod

KafkaStructuredStreamingDemo.scala

Nasıl gönderilir

# 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

Spark StructuredStreaming LogHub örneği (Scala)

Örnek kod

LoghubStructuredStreamingDemo.scala

Nasıl gönderilir

# 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

PySpark kullanarak MaxCompute tablosundan veri okuma veya MaxCompute tablosuna veri yazma örneği (Python)

Örnek kod

spark_sql.py

Nasıl gönderilir

# 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

PySpark kullanarak OSS'e veri yazma örneği (Python)

Örnek kod

spark_oss.py

Nasıl gönderilir

# 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.

Spark SQL örneği (Java)

Spark SQL Java örnek kodu hakkında daha fazla bilgi için bkz. JavaSparkSQL.java.

MaxCompute'dan veri okuma ve HBase'e veri yazma örneği

MaxCompute'dan veri okumak ve bu verileri HBase'e yazmak için IntelliJ IDEA kullanarak kod yazın.

Örnek kod

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()
  }

  }
}

Nasıl gönderilir: Örnek kodu göndermek ve çalıştırmak için IntelliJ IDEA'yı kullanın. Daha fazla bilgi için bkz. Running modes of Spark on MaxCompute.

OSS nesnelerinden veri okuma ve OSS nesnelerine veri yazma örnekleri

OSS nesnelerinden veri okumak ve bu verileri OSS nesnelerine yazmak için IntelliJ IDEA veya DataWorks'u kullanın.

Örnek kod

Örnek 1: Yerel mod için örnek kod

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()
    }
  }
}

Not

Örnek kodu çalıştırmadan önce hadoop-fs-oss bağımlılığının eklendiğini kontrol edin. Bağımlılık eklenmemişse bir hata döndürülür.

Örnek 2: Yerel mod için örnek kod

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()
    }
  }
}

Örnek 3: Küme modu için örnek kod

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()
    }
  }
}

Nasıl gönderilir
- Yerel mod için kodu geliştirmek, test etmek ve göndermek üzere IntelliJ IDEA'yı kullanın. Daha fazla bilgi için bkz. Running modes of Spark on MaxCompute.
- DataWorks konsolunda kodu göndermek ve çalıştırmak için bir ODPS Spark düğümü kullanın. Daha fazla bilgi için bkz. Develop an ODPS Spark task.

MaxCompute'dan veri okuma ve verileri OSS'e yazma örneği

MaxCompute'dan veri okumak ve bu verileri OSS'e yazmak için IntelliJ IDEA veya DataWorks'u kullanın.

Örnek kod

Yerel mod için örnek kod

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()
    }

  }
}

Küme modu için örnek kod

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()
    }

  }
}

Nasıl gönderilir
- Yerel mod için kodu geliştirmek, test etmek ve göndermek üzere IntelliJ IDEA'yı kullanın.
- DataWorks konsolunda kodu göndermek ve çalıştırmak için bir ODPS Spark düğümü kullanın. Daha fazla bilgi için bkz. Develop an ODPS Spark task.
Not
Spark geliştirme ortamının nasıl yapılandırılacağı hakkında daha fazla bilgi için bkz. Running modes of Spark on MaxCompute.