Spark Core入门1【Spark集群安装、高可用、任务执行流程、使用Scala/Java/Lambda编写Spark WordCount】

tar -zxvf spark-2.3.3-bin-hadoop2.7.tgz -C /root/apps/spark

cd /root/apps/spark

cd conf/
mv spark-env.sh.template spark-env.sh
vi spark-env.sh

export JAVA_HOME=/usr/local/jdk1.8
export SPARK_MASTER_IP=hdp-01
export SPARK_MASTER_PORT=7077

mv slaves.template slaves
vi slaves

hdp-02
hdp-03

for i in {2,3}; do scp -r /root/apps/spark/ hdp-0$i:/root/apps; done

/root/apps/spark/sbin/start-all.sh

export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=hdp-01:2181,hdp-02:2181,hdp-03:2181 -Dspark.deploy.zookeeper.dir=/spark"

#除此之外还可以修改Worker的核数(线程数，不能超过实际物理机器的线程)、内存大小等配置
#配置文件上有示例,直接抄即可
export SPARK_WORKER_CORES=8
export SPARK_WORKER_MEMORY=2g

/root/apps/spark-2.3.3-bin-hadoop2.7/bin/spark-submit \
--master spark://hdp-01:7077,spark://hdp-02:7077 \
--class org.apache.spark.examples.SparkPi \
--executor-memory 2048mb \
--total-executor-cores 24 \
/root/apps/spark-2.3.3-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.3.3.jar \
10000

/**
解析:
第一行是指通过spark提交任务的客户端spark-submit
第二行是指定master所在的机器 底层是rpc通信协议 spark://主机名或ip地址:7077  7077是rpc通信端口,而8080是http外部访问端口,需要区分开来。提交任务可以指定多个master地址，目的是为了提交任务高可用
第三行是指执行哪一个类 全路径类名,官方自带的蒙特卡罗求Pi样例(底层是通过反射执行)
第四、五行是指执行的内存大小,cpu核数(实际上这里的核数是执行的线程数)
第六行是指该样例所在的jar包位置   2.11为scala版本  2.3.3为spark版本
第六行是指采样的次数,采样次数越多,求Pi越精确
*/

/root/apps/spark-2.3.3-bin-hadoop2.7/bin/spark-shell

/root/apps/spark-2.3.3-bin-hadoop2.7/bin/spark-shell  \
--master spark://hdp-01:7077 \
--executor-memory 2g \
--total-executor-cores 2

scala> sc.textFile("/root/w.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

scala> sc.textFile("hdfs://hdp-01:9000/wordcount").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect
res1: Array[(String, Int)] = Array((scala,1), (hello,3), (java,1), (spark,2), (hi,2), (dianxin,2))

scala> sc.textFile("hdfs://hdp-01:9000/wordcount").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).sortBy(_._2).collect
res2: Array[(String, Int)] = Array((scala,1), (java,1), (spark,2), (hi,2), (dianxin,2), (hello,3))

scala> sc.textFile("hdfs://hdp-01:9000/wordcount").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).sortBy(_._2,false).collect 
res3: Array[(String, Int)] = Array((hello,3), (spark,2), (hi,2), (dianxin,2), (scala,1), (java,1))

<?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.itcats</groupId>
    <artifactId>spark-wordcount</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <scala.version>2.11.8</scala.version>
        <spark.version>2.2.0</spark.version>
        <hadoop.version>2.6.5</hadoop.version>
        <encoding>UTF-8</encoding>
    </properties>

    <dependencies>
        <!-- 导入scala的依赖 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>

        <!-- 导入spark的依赖 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <!-- 指定hadoop-client API的版本 -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>

    </dependencies>

    <build>
        <pluginManagement>
            <plugins>
                <!-- 编译scala的插件 -->
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                </plugin>
                <!-- 编译java的插件 -->
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.5.1</version>
                </plugin>
            </plugins>
        </pluginManagement>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>compile</phase>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>


            <!-- 打jar插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.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>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ScalaWordCount {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConfig配置对象,配置Spark应用程序的名字
    val conf: SparkConf = new SparkConf()
    conf.setAppName("scalaWordCount")
    //2.创建Spark执行入口————SparkContext Driver和Master通信就是通过SparkContext进行通信
    val sc = new SparkContext(conf)
    //3.指定以后从哪读取数据创建RDD(弹性分布式数据集)
    val lines: RDD[String] = sc.textFile(args(0)) //返回的结果是读取的一行行文件数据集
    //4.切分压平
    val words: RDD[String] = lines.flatMap(_.split(" "))
    //5.将单词和1组合在一起变成元组
    val wordWithOne: RDD[(String, Int)] = words.map((_,1))
    //6.按照key进行聚合
    val reduced: RDD[(String, Int)] = wordWithOne.reduceByKey(_ + _)
    //7.排序
    val sortReduced = reduced.sortBy(_._2, false)   //_为元组(key,出现次数),按出现次数降序排列
    //8.将结果存入HDFS中
    sortReduced.saveAsTextFile(args(1))
    //9.释放资源sc
    sc.stop()
  }
}

[root@hdp-01 bin]# ./spark-submit --master spark://hdp-01:7077 --class cn.itcats.spark.ScalaWordCount /root/spark-wordcount-1.0-SNAPSHOT.jar hdfs://hdp-01:9000/wordcount hdfs://hdp-01:9000/wordcount_res

Exception in thread "main" org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory hdfs://hdp-01:9000/wordcount already exists

[root@hdp-01 bin]# hadoop fs -ls /wordcount_res
Found 3 items
-rw-r--r--   3 root supergroup          0 2019-04-28 21:42 /wordcount_res/_SUCCESS
-rw-r--r--   3 root supergroup         10 2019-04-28 21:42 /wordcount_res/part-00000
-rw-r--r--   3 root supergroup         48 2019-04-28 21:42 /wordcount_res/part-00001

[root@hdp-01 bin]# hadoop fs -cat /wordcount_res/part-00000
(hello,3)
[root@hdp-01 bin]# hadoop fs -cat /wordcount_res/part-00001
(spark,2)
(hi,2)
(dianxin,2)
(scala,1)
(java,1)

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class JavaWordCount {
    public static void main(String[] args) {
        //1.创建SparkContext对象
        SparkConf sparkConf = new SparkConf().setAppName("javaWordCount");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        //2.指定以后从哪读数据
        JavaRDD<String> lines = sc.textFile(args[0]);
        //3.读取的数据为一行行的RDD数据集  切分压平   输入为String类型  输出也为String类型
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String lines) throws Exception {
                //lines.split(" ")返回类型为数组类型,需要返回Iterator类型
                return Arrays.asList(lines.split(" ")).iterator();
            }
        });
        //4.将words组装为元组类型  传入String的words 返回元组 需要调用mapToPair
        JavaPairRDD<String, Integer> wordWithOne = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word, 1);
            }
        });
        //5.将key相同的元素聚合在一起
        JavaPairRDD<String, Integer> reduced = wordWithOne.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        //6.对结果进行排序  发现只有sortByKey  所以应该将Tuple中的键值对换位置,调用mapToPair方法
        JavaPairRDD<Integer, String> swaped = reduced.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {
                //return new Tuple2<Integer, String>(tuple._2, tuple._1);
                return tuple.swap();
            }
        });
        //排序后的结果  (次数, Key)
        JavaPairRDD<Integer, String> sorted = swaped.sortByKey(false);
        //再换回 (Key, 次数的顺序)
        JavaPairRDD<String, Integer> res = sorted.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple2) throws Exception {
                return tuple2.swap();
            }
        });
        //7.将结果保存到HDFS中
        res.saveAsTextFile(args[1]);
        //8.关闭sparkContext资源
        sc.stop();
    }
}

[root@hdp-01 bin]# ./spark-submit --master spark://hdp-01:7077 --class cn.itcats.spark.JavaWordCount /root/spark-wordcount-1.0-SNAPSHOT.jar hdfs://hdp-01:9000/wordcount hdfs://hdp-01:9000/wordcount_res

[root@hdp-01 bin]# hadoop fs -ls /wordcount_res
Found 3 items
-rw-r--r--   3 root supergroup          0 2019-04-28 21:42 /wordcount_res/_SUCCESS
-rw-r--r--   3 root supergroup         10 2019-04-28 21:42 /wordcount_res/part-00000
-rw-r--r--   3 root supergroup         48 2019-04-28 21:42 /wordcount_res/part-00001

[root@hdp-01 bin]# hadoop fs -cat /wordcount_res/part-00000
(hello,3)
[root@hdp-01 bin]# hadoop fs -cat /wordcount_res/part-00001
(spark,2)
(hi,2)
(dianxin,2)
(scala,1)
(java,1)

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

public class LambdaWordCount {
    public static void main(String[] args) {
        //1.创建SparkContext对象
        SparkConf sparkConf = new SparkConf().setAppName("lambdaWordCount");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        //2.指定以后从哪读数据
        JavaRDD<String> lines = sc.textFile(args[0]);
        //3.读取的数据为一行行的RDD数据集  切分压平
        JavaRDD<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
        //4.将words组装为元组类型 w为words中的每个单词
        JavaPairRDD<String, Integer> wordWithOne = words.mapToPair(x -> new Tuple2<String, Integer>(x, 1));
        //5.根据Key进行聚合
        JavaPairRDD<String, Integer> reduced = wordWithOne.reduceByKey((m, n) -> m + n);
        //6.调整顺序  (次数,Key)
        JavaPairRDD<Integer, String> swaped1 = reduced.mapToPair(x -> x.swap());
        //7.排序
        JavaPairRDD<Integer, String> sorted = swaped1.sortByKey(false);
        //7.调整顺序 (Key,次数)
        JavaPairRDD<Integer, String> swaped2 = reduced.mapToPair(x -> x.swap());
        //8.将结果保存到HDFS中
        swaped2.saveAsTextFile(args[1]);
        //9.关闭资源
        sc.stop();
    }
}

//1.创建SparkConfig配置对象,配置Spark应用程序的名字
    //2.local为本地单线程执行  local[4]为本地4线程执行   local[*]本地多少线程就多少线程执行
    val conf: SparkConf = new SparkConf().setAppName("sparkWordCount").setMaster("local[4]")