Spark技术架构的组件功能及WordCount的实现

1. spark有几种部署方式，各有什么特点？

local(本地模式)：常用于本地开发测试，本地还分为local单线程和local-cluster多线程;

standalone(集群模式)：典型的Mater/slave模式，不过也能看出Master是有单点故障的；Spark支持ZooKeeper来实现 HA

on yarn(集群模式)： 运行在 yarn 资源管理器框架之上，由 yarn 负责资源管理，Spark 负责任务调度和计算

on mesos(集群模式)： 运行在 mesos 资源管理器框架之上，由 mesos 负责资源管理，Spark 负责任务调度和计算

on cloud(集群模式)：比如 AWS 的 EC2，使用这个模式能很方便的访问 Amazon的 S3;Spark 支持多种分布式存储系统：HDFS 和 S3

 

2. spark技术栈有哪些组件，每个组件有什么功能，处理什么应用场景？

1）Spark core：是其它组件的基础，spark的内核，主要包含：有向循环图、RDD、Lingage、Cache、broadcast等，并封装了底层通讯框架，是Spark的基础。

2）SparkStreaming是一个对实时数据流进行高通量、容错处理的流式处理系统，可以对多种数据源（如Kafka、Flume、Twitter、Zero和TCP 套接字）进行类似Map、Reduce和Join等复杂操作，将流式计算分解成一系列短小的批处理作业。

3）Spark sql：Shark是SparkSQL的前身，Spark SQL的一个重要特点是其能够统一处理关系表和RDD，使得开发人员可以轻松地使用SQL命令进行外部查询，同时进行更复杂的数据分析。

4）BlinkDB ：是一个用于在海量数据上运行交互式 SQL 查询的大规模并行查询引擎，它允许用户通过权衡数据精度来提升查询响应时间，其数据的精度被控制在允许的误差范围内。

5）MLBase是Spark生态圈的一部分专注于机器学习，让机器学习的门槛更低，让一些可能并不了解机器学习的用户也能方便地使用MLbase。MLBase分为四部分：MLlib、MLI、ML Optimizer和MLRuntime。

6）GraphX用于Spark中图和图并行计算;

 

3. Spark统计wordcount思路及实现。

1)        创建配置对象,设置AppName,Master的属性值

2)        创建spark的入口对象SparkContext

3)        调用flatMap将每行的数据转换为List[String]的集合类型

4)        将List中的元素使用map方法映射为一一对应的元组

5)        对元组数据按聚合函数进行聚合,得到统计结果

object WordCount {
  def main(args: Array[String]): Unit = {
   new SparkContext("local", "wordcount", new SparkConf().set("log4j.rootCategory", "WARN, console"))context.textFile("data/words.txt").flatMap(_.split(" +")).map((_, 1)).reduceByKey(_ + _).foreach(println)
  }
}

  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 org.apache.spark.api.java.function.VoidFunction;

  import scala.Tuple2;

  

  import java.util.Arrays;

  import java.util.Iterator;

  

  

  public class WordCount {

    public static void main(String[] args) {

        JavaSparkContext context = new JavaSparkContext("local", "wordcount", new SparkConf());

        context.setLogLevel("ERROR");


        SparkConf conf = new SparkConf();

        conf.setMaster("local");

        conf.setAppName("wc");

        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> lines = sc.textFile("./data/words");

        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

            @Override

            public Iterator<String> call(String line) throws Exception {

                return Arrays.asList(line.split(" ")).iterator();

            }

        });

        JavaPairRDD<String, Integer> pairWords = words.mapToPair(new PairFunction<String, String, Integer>() {

            @Override

            public Tuple2<String, Integer> call(String s) throws Exception {

                return new Tuple2<>(s, 1);

            }

        });

        JavaPairRDD<String, Integer> result = pairWords.reduceByKey(new Function2<Integer, Integer, Integer>() {

            @Override

            public Integer call(Integer v1, Integer v2) throws Exception {

                return v1 + v2;

            }

        });

        result.foreach(new VoidFunction<Tuple2<String, Integer>>() {

            @Override

            public void call(Tuple2<String, Integer> tp) throws Exception {

                System.out.println();

            }

        });

        sc.stop();

    }

}