文档地址
http://mashibing.com/wiki/Spark
上传文件
解压
#cd training
#tar -xvf spark-2.1.0-bin-hadoop2.7.tgz
# rm -rf spark-2.1.0-bin-hadoop2.7.tgz
#mv spark-2.1.0-bin-hadoop2.7 spark
#cd spark
#vi /etc/profile
#./bin/spark-submit --class org.apache.spark.examples.SparkPi ./examples/jars/spark-examples_2.11-2.1.0.jar 10000
浏览器：
192.168.56.200:4040
当spark完成之后，页面就拒绝访问了。
使用spark-shell进行交互式提交
#./bin/spark-shell
#vi hello.txt
hello java
hello java c java c
hello java
hello java
hello c++
hello python
hello c++
hello java
scala> sc
res0: org.apache.spark.SparkContext = org.apache.spark.SparkContext@77648321
scala> sc.textFile("/root/hello.txt")
res1: org.apache.spark.rdd.RDD[String] = /root/hello.txt MapPartitionsRDD[1] at textFile at <console>:25
scala> val lineRdd = sc.textFile("/root/hello.txt")
lineRdd: org.apache.spark.rdd.RDD[String] = /root/hello.txt MapPartitionsRDD[3] at textFile at <console>:24
scala> lineRDD.foreach(println)
hello java
hello java c java c
hello java
hello java
hello c++
hello python
hello c++
hello java
scala> lineRDD.collect
res6: Array[String] = Array(hello java, hello java c java c, hello java, hello java, hello c++, hello python, hello c++, hello java, "")
scala> val wordRDD = lineRDD.map(line =>line.split(" "))
wordRDD: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[6] at map at <console>:26
scala> wordRDD.collect
res7: Array[Array[String]] = Array(Array(hello, java), Array(hello, java, c, java, c), Array(hello, java), Array(hello, java), Array(hello, c++), Array(hello, python), Array(hello, c++), Array(hello, java), Array(""))
scala> val wordRDD = lineRDD.flatMap(line =>line.split(" "))
wordRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[7] at flatMap at <console>:26
scala> wordRDD.collect
res8: Array[String] = Array(hello, java, hello, java, c, java, c, hello, java, hello, java, hello, c++, hello, python, hello, c++, hello, java, "")
scala> wordRDD.foreach(println)
hello
java
hello
java
c
java
c
hello
java
hello
java
hello
c++
hello
python
hello
c++
hello
java
scala> val wordCountRDD = wordRDD.map(word => (word,1))
wordCountRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[8] at map at <console>:28
scala> wordCountRDD.collect
res10: Array[(String, Int)] = Array((hello,1), (java,1), (hello,1), (java,1), (c,1), (java,1), (c,1), (hello,1), (java,1), (hello,1), (java,1), (hello,1), (c++,1), (hello,1), (python,1), (hello,1), (c++,1), (hello,1), (java,1), ("",1))
scala> val resultRDD = wordCountRDD.reduceByKey((x,y) =>x + y)
resultRDD: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[9] at reduceByKey at <console>:30
scala> resultRDD.collect
res11: Array[(String, Int)] = Array((c++,2), (python,1), ("",1), (hello,8), (java,6), (c,2))
scala> val orderedRDD = resultRDD.sortByKey()
orderedRDD: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[10] at sortByKey at <console>:32
scala> orderedRDD.collect
res12: Array[(String, Int)] = Array(("",1), (c,2), (c++,2), (hello,8), (java,6), (python,1))
scala> sc.textFile("/root/hello.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).sortByKey(false).saveAsTextFile("/root/result")
#ll
total 839228
-rw-------. 1 root root 1020 Dec 1 03:12 anaconda-ks.cfg
-rw-r--r--. 1 root root 149756462 Dec 7 02:07 apache-hive-2.1.1-bin.tar.gz
-rw-r--r--. 1 root root 214092195 Dec 7 02:06 hadoop-2.7.3.tar.gz
-rw-r--r--. 1 root root 104659474 Dec 7 02:07 hbase-1.2.6-bin.tar.gz
-rw-r--r--. 1 root root 98 Dec 7 06:21 hello.txt
-rw-r--r--. 1 root root 160162581 Dec 7 02:06 jdk-8u91-linux-x64.rpm
drwxr-xr-x. 2 root root 80 Dec 7 06:51 result
-rw-r--r--. 1 root root 195636829 Dec 7 03:00 spark-2.1.0-bin-hadoop2.7.tgz
-rw-r--r--. 1 root root 35042811 Dec 7 02:06 zookeeper-3.4.10.tar.gz
#cd result
#ll
total 4
-rw-r--r--. 1 root root 49 Dec 7 06:51 part-00000
-rw-r--r--. 1 root root 0 Dec 7 06:51 _SUCCESS
# more part-00000
python,1)
(java,6)
(hello,8)
(c++,2)
(c,2)
(,1)
