第一天：spark基础

[atguigu@hadoop102 sorfware]$ tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz -C /opt/module/
[atguigu@hadoop102 module]$ mv spark-2.1.1-bin-hadoop2.7 spark

[atguigu@hadoop102 spark]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--executor-memory 1G --total-executor-cores 2 \
./examples/jars/spark-examples_2.11-2.1.1.jar 100
(spark版本不同example不同)

结果：
Pi is roughly 3.1419875141987514

bin/spark-submit \
--class <main-class> --master <master-url> \
--deploy-mode <deploy-mode> --conf <key>=<value> \
... # other options
<application-jar> \

1.txt
hello word
hello scala
2.txt
hello spark

bin/spark-shell
...
Spark context Web UI available at http://10.0.0.153:4040
Spark context available as 'sc' (master = local[*], app id = local-1591940789459).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.2.0
      /_/
         
Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_152)
Type in expressions to have them evaluated.
Type :help for more information.

scala> 
-----
[atguigu@hadoop102 spark]$ jps （查看Java进程）
3627 SparkSubmit
4047 Jps

mv slaves.template slaves
vim slaves
hadoop102 
hadoop103
hadoop104

mv spark-env.sh.template spark-env.sh
SPARK_MASTER_HOST=hadoop102
SPARK_MASTER_PORT=7077
也可以百度配置写别的配置参数哦

xsync spark/

spark/sbin/start-all.sh

util.sh (自写shell脚本)
================atguigu@hadoop102================
3330 Jps
3238 Worker
3163 Master
================atguigu@hadoop103================
2966 Jps
2908 Worker
================atguigu@hadoop104================
2978 Worker
3036 Jps
ssh kg@IP  jps  (前提配置好无秘钥登陆)

#!/bin/bash
#1 获取输入参数个数，如果没有参数，直接退出
pcount=$#
if((pcount==0)); then
echo no args;
exit;
fi

#2 获取文件名称
p1=$1
fname=`basename $p1` # 获得文件名
echo fname=$fname

#3 获取上级目录到绝对路径
pdir=`cd -P $(dirname $p1); pwd` # -P 是进入到软链接 实际路径
echo pdir=$pdir

#4 获取当前用户名称
user=`whoami`

#5 循环  一般就是在102上进行修改 然后运行这个脚本实现 103 跟104的同步
for((host=103; host<105; host++)); do
        echo ------------------- hadoop$host --------------
        rsync -rvl $pdir/$fname $user@hadoop$host:$pdir
done

export JAVA_HOME=/usr/local/java/jdk1.8.0_152

bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://hadoop102:7077 \
--executor-memory 1G --total-executor-cores 2 ./examples/jars/spark-examples_2.11-2.1.1.jar 100

spark-shell --master `spark://hadoop102:7077` --executor-memory 1g --total-executor-cores 2

mv spark-defaults.conf.template spark-defaults.conf
开启Log
vi spark-defaults.conf
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://hadoop102:9000/directory

vi spark-env.sh
export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080 
-Dspark.history.retainedApplications=30 
-Dspark.history.fs.logDirectory=hdfs://hadoop102:9000/directory"

xsync spark-defaults.conf
xsync spark-env.sh 

sbin/start-history-server.sh 

 bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://hadoop102:7077 \
--executor-memory 1G --total-executor-cores 2 ./examples/jars/spark-examples_2.11-2.1.1.jar 100

[atguigu@hadoop102 hadoop]$ vi yarn-site.xml

<!--是否启动一个线程检查每个任务正使用的物理内存量，如果任务超出分配值，则直接将其杀掉，默认是true -->
<property>
  <name>yarn.nodemanager.pmem-check-enabled</name>
  <value>false</value>
</property>
<!--是否启动一个线程检查每个任务正使用的虚拟内存量，如果任务超出分配值，则直接将其杀掉，默认是true -->
<property>
  <name>yarn.nodemanager.vmem-check-enabled</name>
  <value>false</value>
</property>

[atguigu@hadoop102 conf]$ vi spark-env.sh
YARN_CONF_DIR=/opt/module/hadoop-2.7.2/etc/hadoop

[atguigu@hadoop102 conf]$ xsync /opt/module/hadoop-2.7.2/etc/hadoop/yarn-site.xml
[atguigu@hadoop102 conf]$ xsync spark-env.sh

[atguigu@hadoop102 spark]$ bin/spark-submit --class org.apache.spark.examples.SparkPi \
--master yarn --deploy-mode client \
./examples/jars/spark-examples_2.11-2.1.1.jar 100

spark.yarn.historyServer.address=hadoop102:18080
spark.history.ui.port=18080

xsxync spark-defaults.conf 

[atguigu@hadoop102 spark]$ sbin/stop-history-server.sh 
stopping org.apache.spark.deploy.history.HistoryServer
[atguigu@hadoop102 spark]$ sbin/start-history-server.sh 
starting org.apache.spark.deploy.history.HistoryServer, logging to /opt/module/spark/logs/spark-atguigu-org.apache.spark.deploy.history.HistoryServer-1-hadoop102.out

[atguigu@hadoop102 spark]$ bin/spark-submit --class org.apache.spark.examples.SparkPi \
--master yarn --deploy-mode client \
./examples/jars/spark-examples_2.11-2.1.1.jar 100

vi spark-env.sh

注释掉如下内容：
#SPARK_MASTER_HOST=hadoop102
#SPARK_MASTER_PORT=7077
添加上如下内容：
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER \
-Dspark.deploy.zookeeper.url=hadoop102:2181,hadoop103:2181,hadoop104:2181 \
-Dspark.deploy.zookeeper.dir=/spark"

xsync spark-env.sh

[atguigu@hadoop102 spark]$ sbin/start-all.sh

[atguigu@hadoop103 spark]$ sbin/start-master.sh

spark-shell --master spark://hadoop102:7077,hadoop103:7077 --executor-memory 2g --total-executor-cores 2

<?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>com.sowhat.demo</groupId>
    <artifactId>sparktest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>
    </dependencies>

    <build>
        <finalName>WordCount</finalName>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <!-- 这个组件让我们不用再 在项目上add frame 选择scala了，可以自动创建 *.scala 文件 -->
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <!-- 主要是定制化打包
                 https://www.cnblogs.com/fnlingnzb-learner/p/10537228.html-->
                <version>3.0.0</version>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>WordCount</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                        <!--将依赖的第三方jar包打包到jar中，这样方便我们发布可执行的jar包。 -->
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <!--名字任意 -->
                        <phase>package</phase>
                        <!-- 绑定到package生命周期阶段上 -->
                        <goals>
                            <goal>single</goal>
                            <!-- 只运行一次 -->
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

package com.sowhat
import org.apache.spark.{SparkConf, SparkContext}
/**
  * @author sowhat
  * @create 2020-06-12 15:55
  */
object WordCount {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("WC")
    val sc = new SparkContext(conf)
    sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).sortBy(_._2).saveAsTextFile(args(1))
    sc.stop()
  }
}

bin/spark-submit --class com.sowhat.WordCount WordCount-jar-with-dependencies.jar ./input ./output

import org.apache.log4j.Logger
import org.apache.log4j.Level
Logger.getLogger("org").setLevel(Level.OFF)
Logger.getLogger("akka").setLevel(Level.OFF)

val conf = new SparkConf().setAppName("WC").setMaster("local[*]")

package com.sowhat

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @author sowhat
  * @create 2020-06-12 15:55
  */
object WordCountLocal {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.OFF)
    Logger.getLogger("akka").setLevel(Level.OFF)

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("WC").setMaster("local[*]")
    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc = new SparkContext(conf)
    //3.使用sc创建RDD并执行相应的transformation和action
    val tuples = sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).sortBy(_._2).collect()
    //4.关闭连接
    tuples.foreach(println)
    sc.stop()
  }
}