Spark Shell怎么使用

這篇文章主要講解了“Spark Shell怎么使用”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“Spark Shell怎么使用”吧！

##使用Spark Shell進行交互分析##

###基礎(chǔ)## Spark的shell提供了一個學(xué)習(xí)API的簡單的方式，和一個強大的交互式分析數(shù)據(jù)的工具。他在Scala（它運行在Java JVM上并且這是一個很好的使用已存在的Java庫的方式）和Python中都是可用的。通過在Spark目錄運行下面的（腳本）開始：

./bin/spark-shell

Spark的最主要的抽象是一個分布式項目的集合，叫做彈性分布式數(shù)據(jù)集（RDD）。RDDs可以通過Hadoop輸入格式（例如HDFS文件）創(chuàng)建，或者通過轉(zhuǎn)換其他RDDs得到。讓我們通過Spark源碼目錄下的文件README的文本創(chuàng)建一個新的RDD：

scala> val textFile = sc.textFile("README.md")

textFile: spark.RDD[String] = spark.MappedRDD@2ee9b6e3

RDDs擁有actions，他們返回值，并且轉(zhuǎn)換，返回一個新的RDDs的指針。讓我們以一些actions開始：

scala> textFile.count() // Number of items in this RDD

res0: Long = 126

scala> textFile.first() // First item in this RDD

res1: String = # Apache Spark

現(xiàn)在，讓我們使用一個轉(zhuǎn)換，我們將使用一個filter轉(zhuǎn)換來返回一個擁有這個文件的項目的子集的新的RDD：

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))

linesWithSpark: spark.RDD[String] = spark.FilteredRDD@7dd4af09

我們可以鏈?zhǔn)降恼{(diào)用轉(zhuǎn)換盒動作：

scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?

res3: Long = 15

###更多的RDD操作### RDD動作和轉(zhuǎn)換可以用于更加復(fù)雜的計算，讓我們說我們想找出包含字?jǐn)?shù)最多的的行：

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)

res4: Long = 15

This first maps a line to an integer value, creating a new RDD. reduce is called on that RDD to find the largest line count. The arguments to map and reduce are Scala function literals (closures), and can use any language feature or Scala/Java library. For example, we can easily call functions declared elsewhere. We’ll use Math.max() function to make this code easier to understand:

scala> import java.lang.Math import java.lang.Math

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b)) res5: Int = 15

一個常見的數(shù)據(jù)流模型是MapReduce，就像大家熟知的Hadoop.Spark可以非常容易的實現(xiàn)MapReduce。

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)

wordCounts: spark.RDD[(String, Int)] = spark.ShuffledAggregatedRDD@71f027b8

現(xiàn)在，我們聯(lián)合flatMap，map和reduceByKey轉(zhuǎn)換來計算文件中每一個單詞出現(xiàn)的次數(shù)作為一個RDD（String，Int）對。要在我們的shell中統(tǒng)計單詞數(shù)，我們可以使用collect動作：

scala> wordCounts.collect()

res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)

###緩存### Spark同樣支持將數(shù)據(jù)集放到集群范圍的內(nèi)存中緩存。當(dāng)數(shù)據(jù)需要重復(fù)訪問時這非常有用。例如，當(dāng)查詢一個小的”熱點“數(shù)據(jù)集或者當(dāng)運行一個迭代算法就像PageRank。作為一個簡單的例子，讓我們將我們的lineWithSpark數(shù)據(jù)集進行緩存。

scala> linesWithSpark.cache() res7: spark.RDD[String] = spark.FilteredRDD@17e51082

scala> linesWithSpark.count() res8: Long = 15

scala> linesWithSpark.count() res9: Long = 15

##獨立應(yīng)用程序## 現(xiàn)在，假設(shè)我們想使用Spark API開發(fā)一個獨立應(yīng)用程序。我們將通過一個簡單的使用Scala（使用SBT），Java（使用Maven）和Python應(yīng)用程序：

這個示例將使用Maven來編譯一個應(yīng)用程序Jar，但是其他任何相似的構(gòu)建工具同樣可以工作。

我們將創(chuàng)建一個非常簡單的Spark應(yīng)用程序，SimpleApp.java。

import org.apache.spark.api.java.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;

public class SimpleApp {
 
   public static void main(String agrs[]){

          String logFile="YOUR_SPARK_HOME/README.md";
          SparkConf conf=new SparkConf().setAppName("Simple Application");
          JavaSparkContext sc=new JavaSparkContext(conf);
          JavaDDD<String> logData=sc.textFile(logFile).cache();

          long numAs=logData.filter(new Function<String,Boolean>(){
				public Boolean call(String s){
 					return s.contains("a");
				}
		   }).count();

		
		  long numBs=logData.filter(new Function<String,Boolean>(){
				public Boolean call(String s){
 					return s.contains("b");
				}
		   }).count();

           
 		   System.out.println("Lines with a:"+numAs+" ,lines with b:"+numBs);

   }

}

這個程序只是計算一個文本文件中包含”a"的行數(shù)和包含“b”的行數(shù)。注意，你必須使用Spark安裝的位置替換掉“YOUR_SPARK_HOME".在Scala的例子中，我們初始化一個SparkContext，盡管這里我們使用一個特殊的JavaSparkContext類來獲得一個Java友好的（SparkContext）。同時我們創(chuàng)建RDDs（使用JavaRDD表示）并且在他們上面運行轉(zhuǎn)換。最后，我們通過創(chuàng)建一個實現(xiàn)spark.api.java.function.Function的類來傳遞函數(shù)給Spark。Spark編程指南更加詳細的描述了他們的不同點。

要構(gòu)建這個項目，我們編寫一個Maven的pox.xml文件，它將Spark作為一個依賴列出。注意，Spark artifacts被標(biāo)記為一個Scala版本。

<project>
  <groupId>edu.berkeley</groupId>
  <artifactId>simple-project</artifactId>
  <modelVersion>4.0.0</modelVersion>
  <name>Simple Project</name>
  <packaging>jar</packaging>
  <version>1.0</version>
  <repositories>
    <repository>
      <id>Akka repository</id>
      <url>http://repo.akka.io/releases</url>
    </repository>
  </repositories>
  <dependencies>
    <dependency> <!-- Spark dependency -->
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.10</artifactId>
      <version>1.0.2</version>
    </dependency>
  </dependencies>
</project>

我們按照規(guī)定的Maven目錄架構(gòu)創(chuàng)建了這些文件：

$ find .
./pom.xml
./src
./src/main
./src/main/java
./src/main/java/SimpleApp.java

現(xiàn)在，我們可以使用Maven打包這個應(yīng)用，并且使用./bin/spark-submit執(zhí)行它。

# Package a jar containing your application
$ mvn package
...
[INFO] Building jar: {..}/{..}/target/simple-project-1.0.jar

# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
  --class "SimpleApp" \
  --master local[4] \
  target/simple-project-1.0.jar
...
Lines with a: 46, Lines with b: 23

##Where to Go from Here## 恭喜你運行了第一個Spark應(yīng)用程序。

• For an in-depth overview of the API, start with the Spark programming guide, or see “Programming Guides” menu for other components.

• For running applications on a cluster, head to the deployment overview.

• Finally, Spark includes several samples in the examples directory (Scala, Java, Python). You can run them as follows:

  For Scala and Java, use run-example: ./bin/run-example SparkPi

  For Python examples, use spark-submit directly: ./bin/spark-submit examples/src/main/python/pi.py

感謝各位的閱讀，以上就是“Spark Shell怎么使用”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對Spark Shell怎么使用這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關(guān)知識點的文章，歡迎關(guān)注！