安裝：
sudo tar -zxf /home/hadoop/下載/spark-3.0.1-bin-hadoop3.2.tgz -C /usr/local/ cd /usr/local sudo mv ./spark-3.0.1-bin-hadoop3.2/ ./spark sudo chown -R hadoop:hadoop ./spark cd spark/bin spark-shell
測試：
words.txt
hello me you her
hello me you
hello me
hello
運行：
scala> val textFile = sc.textFile("file:///home/hadoop/下載/words.txt") scala> val counts = textFile.flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _)
scala> counts.collect
配置集群：（Standalone-獨立集群）
master
slave1（worker）
slave2（worker）
slave3（worker）
配置slaves/workers
進入配置目錄
cd /usr/local/spark/conf
cp slaves.template slaves
vim slaves
內容如下：
配置master
cp spark-env.sh.template spark-env.sh
vim spark-env.sh
內容如下：
在最下面寫入：
## 設置JAVA安裝目錄 JAVA_HOME=/usr/share/java/jdk1.8.0_261 ## HADOOP軟件配置文件目錄，讀取HDFS上文件和運行Spark在YARN集群時需要,先提前配上 HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop YARN_CONF_DIR=/usr/local/hadoop/etc/hadoop ## 指定spark老大Master的IP和提交任務的通信端口 SPARK_MASTER_HOST=master SPARK_MASTER_PORT=7077 SPARK_MASTER_WEBUI_PORT=8080 SPARK_WORKER_CORES=1 SPARK_WORKER_MEMORY=2g
分發
cd /usr/local sudo scp -r spark hadoop@slave1:$PWD sudo scp -r spark hadoop@slave2:$PWD sudo scp -r spark hadoop@slave3:$PWD
若出現：
則在目標主機上執行：
sudo chmod 777 /usr/local/
再次執行分發命令即可
測試
集群啟動和停止
在主節點上啟動spark集群
cd /usr/local/spark/sbin
./start-all.sh
在主節點上停止spark集群
./stop-all.sh
jps查看進程
master：
slave1
訪問：
http://master:8080/
啟動spark-shell
cd /usr/local/spark/bin spark-shell --master spark://master:7077
提交WordCount任務
注意:上傳文件到hdfs方便worker讀取
上傳文件到hdfs
hadoop fs -put /home/hadoop/下載/words.txt /wordcount/input/words.txt
在shell上：
scala> val textFile = sc.textFile("hdfs://master:9000/wordcount/input/words.txt"
scala> val counts = textFile.flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _)
scala> counts.collect
將結果寫到hdfs文件系統：
counts.saveAsTextFile("hdfs://master:9000/wordcount/output")
查看spark任務web-ui
http://master:4040/
總結:
spark: 4040 任務運行web-ui界面端口
spark: 8080 spark集群web-ui界面端口
spark: 7077 spark提交任務時的通信端口
啟動zk(每台機器上)
cd /usr/local/zookeeper/bin/
./zkServer.sh start
修改配置
cd /usr/local/spark/conf
vim spark-env.sh
注釋
#SPARK_MASTER_HOST=master
修改端口為8888
增加：
SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=master:2181,slave1:2181,slave2:2181,slave3:2181 -Dspark.deploy.zookeeper.dir=/spark"
分發配置
scp -r spark-env.sh hadoop@slave1:$PWD scp -r spark-env.sh hadoop@slave2:$PWD scp -r spark-env.sh hadoop@slave3:$PWD
測試：
在master上啟動Spark集群執行：
cd /usr/local/spark/sbin
./start-all.sh
在slave1上再單獨只起個master：
cd /usr/local/spark/sbin
./start-master.sh
查看：
master：
模擬node1宕機
jps
kill -9 10445
在實際開發中, 大數據任務都有統一的資源管理和任務調度工具來進行管理! ---Yarn使用的最多!
因為它成熟穩定, 支持多種調度策略:FIFO/Capcity/Fair
可以使用Yarn調度管理MR/Hive/Spark/Flink
cd /usr/local/spark/sbin
stop-all.sh
cd /usr/local/hadoop/etc/hadoop
vim yarn-site.xml
內容如下：
<configuration>
<property>
<name>yarn.resourcemanager.address</name>
<value>master:18040</value>
</property>
<property>
<name>yarn.resourcemanager.scheduler.address</name>
<value>master:18030</value>
</property>
<property>
<name>yarn.resourcemanager.webapp.address</name>
<value>master:18088</value>
</property>
<property>
<name>yarn.resourcemanager.resource-tracker.address</name>
<value>master:18025</value>
</property>
<property>
<name>yarn.resourcemanager.admin.address</name>
<value>master:18141</value>
</property>
<property>
<name>yarn.nodemanager.aux-services</name>
<value>mapreduce_shuffle</value>
</property>
<property>
<name>yarn.resourcemanager.hostname</name>
<value>master</value>
</property>
<property>
<name>yarn.nodemanager.auxservices.mapreduce.shuffle.class</name>
<value>org.apache.hadoop.mapred.ShuffleHandler</value>
</property>
<!-- 設置yarn集群的內存分配方案 -->
<property>
<name>yarn.nodemanager.resource.memory-mb</name>
<value>20480</value>
</property>
<property>
<name>yarn.scheduler.minimum-allocation-mb</name>
<value>2048</value>
</property>
<property>
<name>yarn.nodemanager.vmem-pmem-ratio</name>
<value>2.1</value>
</property>
<!-- 開啟日志聚合功能 -->
<property>
<name>yarn.log-aggregation-enable</name>
<value>true</value>
</property>
<!-- 設置聚合日志在hdfs上的保存時間 -->
<property>
<name>yarn.log-aggregation.retain-seconds</name>
<value>604800</value>
</property>
<!-- 設置yarn歷史服務器地址 -->
<property>
<name>yarn.log.server.url</name>
<value>http://master:19888/jobhistory/logs</value>
</property>
<!-- 關閉yarn內存檢查 -->
<property>
<name>yarn.nodemanager.pmem-check-enabled</name>
<value>false</value>
</property>
<property>
<name>yarn.nodemanager.vmem-check-enabled</name>
<value>false</value>
</property>
<property>
<name>yarn.application.classpath</name>
<value>/usr/local/hadoop/etc/hadoop:/usr/local/hadoop/share/hadoop/common/lib/*:/usr/local/hadoop/share/hadoop/common/*:/usr/local/hadoop/share/hadoop/hdfs:/usr/local/hadoop/share/hadoop/hdfs/lib/*:/usr/local/hadoop/share/hadoop/hdfs/*:/usr/local/hadoop/share/hadoop/mapreduce/lib/*:/usr/local/hadoop/share/hadoop/mapreduce/*:/usr/local/hadoop/share/hadoop/yarn:/usr/local/hadoop/share/hadoop/yarn/lib/*:/usr/local/hadoop/share/hadoop/yarn/*</value>
</property>
</configuration>
分發：
scp -r yarn-site.xml hadoop@slave1:$PWD
scp -r yarn-site.xml hadoop@slave3:$PWD
scp -r yarn-site.xml hadoop@slave2:$PWD
cd /usr/local/spark/conf cp spark-defaults.conf.template spark-defaults.conf vim spark-defaults.conf
增加：
spark.eventLog.enabled true spark.eventLog.dir hdfs://master:9000/sparklog/ spark.eventLog.compress true spark.yarn.historyServer.address master:18080
修改spark-env.sh
vim spark-env.sh
增加：
## 配置spark歷史日志存儲地址 SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://master:9000/sparklog/ -Dspark.history.fs.cleaner.enabled=true"
注意:sparklog需要手動創建
hadoop fs -mkdir -p /sparklog
修改日志級別
cd /usr/local/spark/conf
cp log4j.properties.template log4j.properties
vim log4j.properties
分發：
scp -r spark-env.sh hadoop@slave1:$PWD scp -r spark-env.sh hadoop@slave2:$PWD scp -r spark-env.sh hadoop@slave3:$PWD scp -r spark-defaults.conf hadoop@slave1:$PWD scp -r spark-defaults.conf hadoop@slave2:$PWD scp -r spark-defaults.conf hadoop@slave3:$PWD scp -r log4j.properties hadoop@slave1:$PWD log4j.properties scp -r log4j.properties hadoop@slave2:$PWD log4j.properties scp -r log4j.properties hadoop@slave3:$PWD log4j.properties
hadoop fs -mkdir -p /spark/jars/
上傳$SPARK_HOME/jars所有jar包到HDFS
hadoop fs -put /usr/local/spark/jars/* /spark/jars/
修改spark-defaults.conf
vim spark-defaults.conf
增加：
spark.yarn.jars hdfs://master:9000/spark/jars/*
分發：
scp -r spark-defaults.conf hadoop@slave1:$PWD scp -r spark-defaults.conf hadoop@slave2:$PWD scp -r spark-defaults.conf hadoop@slave3:$PWD
啟動HDFS和YARN服務
start-dfs.sh
start-yarn.sh
或
start-all.sh
啟動MRHistoryServer服務
mr-jobhistory-daemon.sh start historyserver
啟動Spark HistoryServer服務
cd /usr/local/spark/sbin
start-history-server.sh
MRHistoryServer服務WEB UI頁面：
Spark HistoryServer服務WEB UI頁面：
1.需要Yarn集群
2.歷史服務器
3.提交任務的的客戶端工具-spark-submit命令
4.待提交的spark任務/程序的字節碼--可以使用示例程序
SPARK_HOME=/usr/local/spark ${SPARK_HOME}/bin/spark-submit \ --master yarn \ --deploy-mode client \ --driver-memory 512m \ --driver-cores 1 \ --executor-memory 512m \ --num-executors 2 \ --executor-cores 1 \ --class org.apache.spark.examples.SparkPi \ ${SPARK_HOME}/examples/jars/spark-examples_2.12-3.0.1.jar \ 10
查看web界面
SPARK_HOME=/usr/local/spark ${SPARK_HOME}/bin/spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 512m \ --executor-memory 512m \ --num-executors 1 \ --class org.apache.spark.examples.SparkPi \ ${SPARK_HOME}/examples/jars/spark-examples_2.12-3.0.1.jar \ 10
查看web界面
Spark程序開發
創建maven項目
添加pom.xml內容
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>cn.itcast</groupId> <artifactId>spark_study_47</artifactId> <version>1.0-SNAPSHOT</version> <repositories> <repository> <id>aliyun</id> <url>http://maven.aliyun.com/nexus/content/groups/public/</url> </repository> <repository> <id>apache</id> <url>https://repository.apache.org/content/repositories/snapshots/</url> </repository> <repository> <id>cloudera</id> <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url> </repository> </repositories> <properties> <encoding>UTF-8</encoding> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <scala.version>2.12.11</scala.version> <spark.version>3.0.1</spark.version> <hadoop.version>2.7.5</hadoop.version> </properties> <dependencies> <!--依賴Scala語言--> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <!--SparkCore依賴--> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12</artifactId> <version>${spark.version}</version> </dependency> <!-- spark-streaming--> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.12</artifactId> <version>${spark.version}</version> </dependency> <!--spark-streaming+Kafka依賴--> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka-0-10_2.12</artifactId> <version>${spark.version}</version> </dependency> <!--SparkSQL依賴--> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.12</artifactId> <version>${spark.version}</version> </dependency> <!--SparkSQL+ Hive依賴--> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-hive_2.12</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-hive-thriftserver_2.12</artifactId> <version>${spark.version}</version> </dependency> <!--StructuredStreaming+Kafka依賴--> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql-kafka-0-10_2.12</artifactId> <version>${spark.version}</version> </dependency> <!-- SparkMlLib機器學習模塊,里面有ALS推薦算法--> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-mllib_2.12</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.7.5</version> </dependency> <dependency> <groupId>com.hankcs</groupId> <artifactId>hanlp</artifactId> <version>portable-1.7.7</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.38</version> </dependency> <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.9.0</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.47</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.2</version> <scope>provided</scope> </dependency> </dependencies> <build> <sourceDirectory>src/main/scala</sourceDirectory> <plugins> <!-- 指定編譯java的插件 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.5.1</version> </plugin> <!-- 指定編譯scala的插件 --> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.2</version> <executions> <execution> <goals> <goal>compile</goal> <goal>testCompile</goal> </goals> <configuration> <args> <arg>-dependencyfile</arg> <arg>${project.build.directory}/.scala_dependencies</arg> </args> </configuration> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-surefire-plugin</artifactId> <version>2.18.1</version> <configuration> <useFile>false</useFile> <disableXmlReport>true</disableXmlReport> <includes> <include>**/*Test.*</include> <include>**/*Suite.*</include> </includes> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.3</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass></mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build> </project>
新建scala文件夾：
本地實現：
代碼：
package cn.itcast.hello import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} /** * Author itcast * Desc 演示Spark入門案例-WordCount */ object WordCount_bak { def main(args: Array[String]): Unit = { //TODO 1.env/准備sc/SparkContext/Spark上下文執行環境 val conf: SparkConf = new SparkConf().setAppName("wc").setMaster("local[*]") val sc: SparkContext = new SparkContext(conf) sc.setLogLevel("WARN") //TODO 2.source/讀取數據 //RDD:A Resilient Distributed Dataset (RDD):彈性分布式數據集,簡單理解為分布式集合!使用起來和普通集合一樣簡單! //RDD[就是一行行的數據] val lines: RDD[String] = sc.textFile("data/input/words.txt") //TODO 3.transformation/數據操作/轉換 //切割:RDD[一個個的單詞] val words: RDD[String] = lines.flatMap(_.split(" ")) //記為1:RDD[(單詞, 1)] val wordAndOnes: RDD[(String, Int)] = words.map((_,1)) //分組聚合:groupBy + mapValues(_.map(_._2).reduce(_+_)) ===>在Spark里面分組+聚合一步搞定:reduceByKey val result: RDD[(String, Int)] = wordAndOnes.reduceByKey(_+_) //TODO 4.sink/輸出 //直接輸出 result.foreach(println) //收集為本地集合再輸出 println(result.collect().toBuffer) //輸出到指定path(可以是文件/夾) result.repartition(1).saveAsTextFile("data/output/result") result.repartition(2).saveAsTextFile("data/output/result2") result.saveAsTextFile("data/output/result3") //為了便於查看Web-UI可以讓程序睡一會 Thread.sleep(1000 * 60) //TODO 5.關閉資源 sc.stop() } }
package cn.itcast.hello import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} /** * Author itcast * Desc 演示Spark入門案例-WordCount-修改代碼使適合在Yarn集群上運行 */ object WordCount { def main(args: Array[String]): Unit = { if(args.length < 2){ println("請指定input和output") System.exit(1)//非0表示非正常退出程序 } //TODO 1.env/准備sc/SparkContext/Spark上下文執行環境 val conf: SparkConf = new SparkConf().setAppName("wc")//.setMaster("local[*]") val sc: SparkContext = new SparkContext(conf) sc.setLogLevel("WARN") //TODO 2.source/讀取數據 //RDD:A Resilient Distributed Dataset (RDD):彈性分布式數據集,簡單理解為分布式集合!使用起來和普通集合一樣簡單! //RDD[就是一行行的數據] val lines: RDD[String] = sc.textFile(args(0))//注意提交任務時需要指定input參數 //TODO 3.transformation/數據操作/轉換 //切割:RDD[一個個的單詞] val words: RDD[String] = lines.flatMap(_.split(" ")) //記為1:RDD[(單詞, 1)] val wordAndOnes: RDD[(String, Int)] = words.map((_,1)) //分組聚合:groupBy + mapValues(_.map(_._2).reduce(_+_)) ===>在Spark里面分組+聚合一步搞定:reduceByKey val result: RDD[(String, Int)] = wordAndOnes.reduceByKey(_+_) //TODO 4.sink/輸出 //直接輸出 //result.foreach(println) //收集為本地集合再輸出 //println(result.collect().toBuffer) //輸出到指定path(可以是文件/夾) //如果涉及到HDFS權限問題不能寫入,需要執行: //hadoop fs -chmod -R 777 / //並添加如下代碼 System.setProperty("HADOOP_USER_NAME", "root") result.repartition(1).saveAsTextFile(args(1))//注意提交任務時需要指定output參數 //為了便於查看Web-UI可以讓程序睡一會 //Thread.sleep(1000 * 60) //TODO 5.關閉資源 sc.stop() } }
打包：
改為wc.jar
上傳到linux上
提交任務
先啟動yarn集群：
start-all.sh
運行：
SPARK_HOME=/usr/local/spark ${SPARK_HOME}/bin/spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 512m \ --executor-memory 512m \ --num-executors 1 \ --class cn.itcast.hello.WordCount \ /home/hadoop/下載/wc.jar \ hdfs://master:9000/wordcount/input/words.txt \ hdfs://master:9000/wordcount/output_2
