Spark Shell怎么使用

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

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

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

./bin/spark-shell

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

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

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

RDDs擁有actions，他們返回值，并且轉換，返回一個新的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

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

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

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

我們可以鏈式的調用轉換盒動作：

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

res3: Long = 15

###更多的RDD操作### RDD動作和轉換可以用于更加復雜的計算，讓我們說我們想找出包含字數最多的的行：

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

一個常見的數據流模型是MapReduce，就像大家熟知的Hadoop.Spark可以非常容易的實現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

現在，我們聯合flatMap，map和reduceByKey轉換來計算文件中每一個單詞出現的次數作為一個RDD（String，Int）對。要在我們的shell中統計單詞數，我們可以使用collect動作：

scala> wordCounts.collect()

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

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

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

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

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

##獨立應用程序## 現在，假設我們想使用Spark API開發一個獨立應用程序。我們將通過一個簡單的使用Scala（使用SBT），Java（使用Maven）和Python應用程序：

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

我們將創建一個非常簡單的Spark應用程序，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"的行數和包含“b”的行數。注意，你必須使用Spark安裝的位置替換掉“YOUR_SPARK_HOME".在Scala的例子中，我們初始化一個SparkContext，盡管這里我們使用一個特殊的JavaSparkContext類來獲得一個Java友好的（SparkContext）。同時我們創建RDDs（使用JavaRDD表示）并且在他們上面運行轉換。最后，我們通過創建一個實現spark.api.java.function.Function的類來傳遞函數給Spark。Spark編程指南更加詳細的描述了他們的不同點。

要構建這個項目，我們編寫一個Maven的pox.xml文件，它將Spark作為一個依賴列出。注意，Spark artifacts被標記為一個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>

我們按照規定的Maven目錄架構創建了這些文件：

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

現在，我們可以使用Maven打包這個應用，并且使用./bin/spark-submit執行它。

# 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應用程序。

• 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怎么使用”的內容了，經過本文的學習后，相信大家對Spark Shell怎么使用這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！