Spark學習筆記——安裝和WordCount

1.去清華的鏡像站點下載文件spark-2.1.0-bin-without-hadoop.tgz，不要下spark-2.1.0-bin-hadoop2.7.tgz

2.把文件解壓到/usr/local目錄下，解壓之后的效果，Hadoop和Spark都在Hadoop用戶下

下面的操作都在Hadoop用戶下

drwxrwxrwx 13 hadoop hadoop 4096 4月   4 11:50 spark-2.1.0-bin-without-hadoop/

 添加Hadoop用戶和用戶組

$ sudo addgroup hadoop
$ sudo adduser --ingroup hadoop hadoop
$ sudo adduser hadoop sudo

 然后修改文件夾的用戶，用戶組以及權限

sudo chown -R hduser:hadoop spark-2.1.0-bin-without-hadoop
sudo chmod 777 hadoop/

 Hadoop文件夾如果權限不對的話，也需要修改

3.在/etc/profile下添加路徑

export SPARK_HOME=/usr/local/spark-2.1.0-bin-without-hadoop
export PATH=${SPARK_HOME}/bin:$PATH

4.還需要修改Spark的配置文件spark-env.sh

cd /usr/local/spark-2.1.0-bin-without-hadoop
cp ./conf/spark-env.sh.template ./conf/spark-env.sh

添加如下

export SPARK_DIST_CLASSPATH=$(/home/lintong/software/apache/hadoop-2.9.1/bin/hadoop classpath)

 （以上可以參考廈門大學林子雨老師的教程——Spark2.1.0入門：Spark的安裝和使用），有些教程坑無數

5.在~/coding/coding/Scala/word-count路徑下准備一個文本文件，比如test.segmented文件

6.在該目錄下，在終端運行 spark-shell

創建一個RDD

scala> val textFile = sc.textFile("file:///home/common/coding/coding/Scala/word-count/test.segmented")

保存RDD成文件

textFile.saveAsTextFile("file:///home/common/coding/coding/Scala/word-count/writeback")

 這時候會發現在文件夾目錄下多了writeback目錄，目錄下是這么幾個文件

 現在，我們建立hdfs文件夾，來把 test.segmented 文件放進我們的hdfs文件夾中

首先，啟動Hadoop的HDFS組件，因為沒有用到MapReduce組件，所以沒有必要啟動MapReducen或者YARN

cd /usr/local/hadoop
./sbin/start-dfs.sh

 在HDFS文件系統中，建立文件夾

./bin/hdfs dfs -mkdir -p /user/hadoop

使用命令查看一下HDFS文件系統中的目錄和文件

在Hadoop文件夾下運行命令

./bin/hdfs dfs -ls .       #或者
./bin/hdfs dfs -ls /user/hadoop

 或者直接

hadoop fs -ls /user/hadoop    #或者
hadoop fs -ls .

 把剛剛的 test.segmented 文件上傳到分布式文件系統HDFS中（放到hadoop用戶目錄下）

hadoop fs -put /home/common/coding/coding/Java/WordCount/input/test.segmented .

 再次查看一下

hadoop@master:~$ hadoop fs -ls /user/hadoop
Found 2 items
drwxr-xr-x   - hadoop supergroup          0 2017-04-03 16:18 /user/hadoop/QuasiMonteCarlo_1491207499210_758373570
-rw-r--r--   1 hadoop supergroup         59 2017-04-03 16:43 /user/hadoop/test.segmented

如果需要刪除

hadoop fs -rm /user/hadoop/test.segmented

查看一個文件的大小

hadoop fs -du -h /logs/xxxx

現在回到 spark-shell 窗口，編寫代碼從HDFS文件系統加載 test.segmented 文件

並打印文件中的第一行內容

scala> val textFile = sc.textFile("hdfs://master:9000/user/hadoop/test.segmented")
textFile: org.apache.spark.rdd.RDD[String] = hdfs://master:9000/user/hadoop/test.segmented MapPartitionsRDD[1] at textFile at <console>:24

scala> textFile.first()
res0: String = aa bb aa

 如果是單機的話，其中下面兩條語句和上面第一條語句是一樣的，但是如果是Hadoop偽分布式或者分布式的話，就不行

val textFile = sc.textFile("/user/hadoop/test.segmented")

 再次把textFile寫回到HDFS文件系統中

textFile.saveAsTextFile("hdfs://master:9000/user/hadoop/writeback")

再次查看

hadoop@master:~$ hadoop fs -ls /user/hadoop
Found 3 items
drwxr-xr-x   - hadoop supergroup          0 2017-04-03 16:18 /user/hadoop/QuasiMonteCarlo_1491207499210_758373570
-rw-r--r--   1 hadoop supergroup         59 2017-04-03 16:43 /user/hadoop/test.segmented
drwxr-xr-x   - hadoop supergroup          0 2017-04-03 17:10 /user/hadoop/writeback

如果進入writeback文件夾中查看的話，可以看到里面的文件的內容和test.segmented中的是一樣的

hadoop@master:~$ hadoop fs -ls /user/hadoop/writeback
Found 3 items
-rw-r--r--   3 hadoop supergroup          0 2017-04-03 17:10 /user/hadoop/writeback/_SUCCESS
-rw-r--r--   3 hadoop supergroup         36 2017-04-03 17:10 /user/hadoop/writeback/part-00000
-rw-r--r--   3 hadoop supergroup         24 2017-04-03 17:10 /user/hadoop/writeback/part-00001

hadoop@master:~$ hadoop fs -cat /user/hadoop/writeback/part-00000
aa bb aa
bb aa aa
cc bb ee
dd ee cc

hadoop@master:~$ hadoop fs -cat /user/hadoop/writeback/part-00001
aa
cc
ee
ff
ff
gg
hh
aa

 

現在進入WordCount階段，再次進入 Spark-shell 中

val textFile = sc.textFile("hdfs://master:9000/user/hadoop/test.segmented")
val wordCount = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCount.collect()

 輸出

res6: Array[(String, Int)] = Array((ee,3), (aa,6), (gg,1), (dd,1), (hh,1), (ff,2), (bb,3), (cc,3))

 

 在spark-shell下面運行成功之后，就需要試着在idea里面建立一個工程來運行這段代碼

在idea下面建立一個Scala的工程，構建的方式選擇是sbt

 

由於本機的Scala的版本是2.11.8

所以在project structure里面設置成2.11.8

接着在build.sbt里面寫

name := "word-count"

version := "1.0"

scalaVersion := "2.11.8"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"

 注意里面的scalaVersion如果是2.11.X的話，sbt就會去拉spark-core_2.11-2.1.0的包

可以去公司的私服nexus里面去看看有沒有這個包

然后在WordCount.scala文件中寫入我們的代碼

注意如果是setMaster("local")的話，需要在/etc/hosts中設置127.0.1.1，然后取消192.168.0.1

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
/**
  * Created by common on 17-4-3.
  */


object WordCount {
  def main(args: Array[String]) {
    val inputFile =  "file:///home/common/coding/coding/Scala/word-count/test.segmented"
    val conf = new SparkConf().setAppName("WordCount").setMaster("local")　　　　#創建一個SparkConf對象來配置應用
　　　　#集群URL：告訴Spark連接到哪個集群，local是單機單線程，無需連接到集群，應用名：在集群管理器的用戶界面方便找到應用
    val sc = new SparkContext(conf)　　　　　　　　#然后基於這SparkConf創建一個SparkContext對象
    val textFile = sc.textFile(inputFile)　　　　#讀取輸入的數據
    val wordCount = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)　　#切分成單詞，轉換成鍵值對並計數
    wordCount.foreach(println)
  }
}

然后在sbt中refresh，進行拉包，拉包的過程是無比緩慢的

拉好了之后運行的結果

和在 spark-shell中運行的結果是一致的

 

在林子雨老師的教程中，Spark2.1.0入門：第一個Spark應用程序：WordCount

最后是將整個應用程序打包成JAR，然后通過 spark-submit 提交到 Spark 中運行

做法是在idea的終端中，對代碼進行打包

common@master:~/coding/coding/Scala/word-count$ sbt package
[info] Loading project definition from /home/common/coding/coding/Scala/word-count/project
[info] Set current project to word-count (in build file:/home/common/coding/coding/Scala/word-count/)
[info] Compiling 1 Scala source to /home/common/coding/coding/Scala/word-count/target/scala-2.11/classes...
[info] Packaging /home/common/coding/coding/Scala/word-count/target/scala-2.11/word-count_2.11-1.0.jar ...
[info] Done packaging.
[success] Total time: 6 s, completed 2017-4-4 18:02:13

 生成的jar包位置在

/home/common/coding/coding/Scala/word-count/target/scala-2.11

 最后通過 spark-submit 運行程序，將jar包通過這個命令提交到 Spark 中運行

common@master:~/coding/coding/Scala/word-count$ spark-submit --class "WordCount"  /home/common/coding/coding/Scala/word-count/target/scala-2.11/word-count_2.11-1.0.jar

 運行結果

 

在執行spark任務的時候，如果遇到

報如下錯誤:
Exception in thread "main" java.lang.Exception: When running with master 'yarn-client' either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment

在spark的配置文件 conf/spark-env.sh 中添加

export HADOOP_HOME=/home/lintong/software/apache/hadoop-2.9.1
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
PATH=$PATH:$HIVE_HOME/bin:$HADOOP_HOME/bin

 

spark集群安裝參考：spark 集群搭建 詳細步驟

主要是配置slave文件和spark-env文件

集群內容spark-env文件，其中xxx是spark web ui的端口

export SPARK_DIST_CLASSPATH=$(/usr/bin/hadoop classpath)

#export JAVA_HOME=/usr/lib/jvm/java-8-oracle
export SCALA_HOME=/home/dl/packages/scala-2.11.8

export SPARK_MASTER_WEBUI_PORT=xxxx

export HADOOP_HOME=/opt/cloudera/parcels/CDH-5.12.0-1.cdh5.12.0.p0.29/lib/hadoop
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export HIVE_HOME=/opt/cloudera/parcels/CDH-5.12.0-1.cdh5.12.0.p0.29/lib/hive
PATH=$PATH:$HIVE_HOME/bin:$HADOOP_HOME/bin