Apache Spark - 部署

Spark 应用程序，使用 spark-submit，是一个用于在集群上部署 Spark 应用程序的 shell 命令。 它通过一个统一的接口使用所有各自的集群管理器。 因此，您不必为每个应用程序配置。

示例

让我们以之前使用 shell 命令的字数统计为例。 在这里，我们考虑与 Spark 应用程序相同的示例。

示例输入

以下文本为输入数据，文件名为in.txt。

people are not as beautiful as they look, 
as they walk or as they talk. 
they are only as beautiful  as they love, 
as they care as they share.

看下面的程序 −

SparkWordCount.scala

import org.apache.spark.SparkContext 
import org.apache.spark.SparkContext._ 
import org.apache.spark._  

object SparkWordCount { 
   def main(args: Array[String]) { 

      val sc = new SparkContext( "local", "Word Count", "/usr/local/spark", Nil, Map(), Map()) 
		
      /* local = master URL; Word Count = application name; */  
      /* /usr/local/spark = Spark Home; Nil = jars; Map = environment */ 
      /* Map = variables to work nodes */ 
      /*creating an inputRDD to read text file (in.txt) through Spark context*/ 
      val input = sc.textFile("in.txt") 
      /* Transform the inputRDD into countRDD */ 
		
      val count = input.flatMap(line ⇒ line.split(" ")) 
      .map(word ⇒ (word, 1)) 
      .reduceByKey(_ + _) 
       
      /* saveAsTextFile method is an action that effects on the RDD */  
      count.saveAsTextFile("outfile") 
      System.out.println("OK"); 
   } 
} 

将上述程序保存到名为 SparkWordCount.scala 的文件中，并将其放置在名为 spark-application 的用户定义目录中。

注意 − 在将 inputRDD 转换为 countRDD 时，我们使用 flatMap() 将行（来自文本文件）标记为单词，使用 map() 方法计算单词频率，使用 reduceByKey() 方法计算每个单词的重复次数。

使用以下步骤提交此申请。 通过终端执行 spark-application 目录中的所有步骤。

步骤 1: 下载 Spark Ja

编译需要 Spark 核心 jar，因此，从以下链接 Spark core jar 下载 spark-core_2.10-1.3.0.jar 并将 jar 文件从下载目录移动到 spark-application 目录 .

步骤 2: 编译程序

使用下面给出的命令编译上面的程序。 此命令应从 spark-application 目录执行。 此处，/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar 是从 Spark 库中获取的 Hadoop 支持 jar。

$ scalac -classpath "spark-core_2.10-1.3.0.jar:/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar" SparkPi.scala

步骤 3: 创建一个 JAR

使用以下命令创建 spark 应用程序的 jar 文件。 这里，wordcount 是 jar 文件的文件名。

jar -cvf wordcount.jar SparkWordCount*.class spark-core_2.10-1.3.0.jar/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar

步骤 4: 提交 spark 申请

使用以下命令提交 spark 应用程序 −

spark-submit --class SparkWordCount --master local wordcount.jar

如果它执行成功，那么你会发现下面给出的输出。 下面输出中的 OK 用于用户识别，这是程序的最后一行。 如果你仔细阅读下面的输出，你会发现不同的东西，比如 −

• successfully started service 'sparkDriver' on port 42954
• MemoryStore started with capacity 267.3 MB
• Started SparkUI at http://192.168.1.217:4040
• Added JAR file:/home/hadoop/piapplication/count.jar
• ResultStage 1 (saveAsTextFile at SparkPi.scala:11) finished in 0.566 s
• Stopped Spark web UI at http://192.168.1.217:4040
• MemoryStore cleared

15/07/08 13:56:04 INFO Slf4jLogger: Slf4jLogger started 
15/07/08 13:56:04 INFO Utils: Successfully started service 'sparkDriver' on port 42954. 
15/07/08 13:56:04 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriver@192.168.1.217:42954] 
15/07/08 13:56:04 INFO MemoryStore: MemoryStore started with capacity 267.3 MB 
15/07/08 13:56:05 INFO HttpServer: Starting HTTP Server 
15/07/08 13:56:05 INFO Utils: Successfully started service 'HTTP file server' on port 56707. 
15/07/08 13:56:06 INFO SparkUI: Started SparkUI at http://192.168.1.217:4040 
15/07/08 13:56:07 INFO SparkContext: Added JAR file:/home/hadoop/piapplication/count.jar at http://192.168.1.217:56707/jars/count.jar with timestamp 1436343967029 
15/07/08 13:56:11 INFO Executor: Adding file:/tmp/spark-45a07b83-42ed-42b3b2c2-823d8d99c5af/userFiles-df4f4c20-a368-4cdd-a2a7-39ed45eb30cf/count.jar to class loader 
15/07/08 13:56:11 INFO HadoopRDD: Input split: file:/home/hadoop/piapplication/in.txt:0+54 
15/07/08 13:56:12 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2001 bytes result sent to driver 
 (MapPartitionsRDD[5] at saveAsTextFile at SparkPi.scala:11), which is now runnable 
15/07/08 13:56:12 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 1 (MapPartitionsRDD[5] at saveAsTextFile at SparkPi.scala:11) 
15/07/08 13:56:13 INFO DAGScheduler: ResultStage 1 (saveAsTextFile at SparkPi.scala:11) finished in 0.566 s 
15/07/08 13:56:13 INFO DAGScheduler: Job 0 finished: saveAsTextFile at SparkPi.scala:11, took 2.892996 s
OK 
15/07/08 13:56:13 INFO SparkContext: Invoking stop() from shutdown hook 
15/07/08 13:56:13 INFO SparkUI: Stopped Spark web UI at http://192.168.1.217:4040 
15/07/08 13:56:13 INFO DAGScheduler: Stopping DAGScheduler 
15/07/08 13:56:14 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped! 
15/07/08 13:56:14 INFO Utils: path = /tmp/spark-45a07b83-42ed-42b3-b2c2823d8d99c5af/blockmgr-ccdda9e3-24f6-491b-b509-3d15a9e05818, already present as root for deletion. 
15/07/08 13:56:14 INFO MemoryStore: MemoryStore cleared 
15/07/08 13:56:14 INFO BlockManager: BlockManager stopped 
15/07/08 13:56:14 INFO BlockManagerMaster: BlockManagerMaster stopped 
15/07/08 13:56:14 INFO SparkContext: Successfully stopped SparkContext 
15/07/08 13:56:14 INFO Utils: Shutdown hook called 
15/07/08 13:56:14 INFO Utils: Deleting directory /tmp/spark-45a07b83-42ed-42b3b2c2-823d8d99c5af 
15/07/08 13:56:14 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!  

步骤 5: 检查输出

程序执行成功后，会在spark-application目录下找到名为outfile的目录。

以下命令用于打开和检查 outfile 目录中的文件列表。

$ cd outfile 
$ ls 
Part-00000 part-00001 _SUCCESS

用于检查 part-00000 文件中的输出的命令是 −

$ cat part-00000 
(people,1) 
(are,2) 
(not,1) 
(as,8) 
(beautiful,2) 
(they, 7) 
(look,1)

用于检查 part-00001 文件中的输出的命令是 −

$ cat part-00001 
(walk, 1) 
(or, 1) 
(talk, 1) 
(only, 1) 
(love, 1) 
(care, 1) 
(share, 1)

阅读以下部分以了解有关"spark-submit"命令的更多信息。

Spark-submit 语法

spark-submit [options] <app jar | python file> [app arguments]

选项