一.大数据简介
大数据是一个很热门的话题，但它是什么时候开始兴起的呢？
大数据【big data】这个词最早在UNIX用户协会的会议上被使用，来自SGI公司的科学家在其文章“大数据与下一代基础架构”【big data and the next wave of infrastress】中用它来描述数据的快速增长。现在一般用4V来表示，及大量【volume】、多样【variety】、快速【velocity】和价值【value】。
二.大数据时代所面临的问题
1.数据的快速增长使快速处理数据成为了空中楼阁。就拿搜索引擎为例，搜索引擎需要把网上的多数网页抓取下来进行更分析，并建立索引，这样在我们搜索一个词的时候，它才能在毫秒级别返回数据。我们假设每个网页的平均大小为20kb，大概有加载的中文网页有200亿个页面，不过考虑压缩的情况，大概总共有400TB的大小。那么一台计算机以30MB/s~35MB/s的速度读取硬盘，大概需要4个多月，这还不包括在这些数据上做处理所花费的时间。因此，通常我们需要一个集群来处理这一类的工作。可能有人说计算机的处理能力在以指数方式提升，这样机器就能处理更多的数据。是的，这是事实，但是这些计算机生产数据的能力也是以指数方式增长。而且，随着可穿戴设备，智能家居等一系列的增加，数据生产的速度会也来越快。有数据表明，90%的数据是最近两年左右产生的。
2.以集群为例，随着数据的逐渐增加，集群的规模也会越来越大，这样集群出现问题的可能性也在大大增加。以1000台的集群为例，一台机器一个月正常工作的可能性为0.999，那么所有1000台一个月都正常的概率为0.999的1000次方，约等于0.37，不到一半。所有，随着数据和机器的增加，处理数据的技术也变得越来越复杂了。为了解决这些问题，各种大数据方案诞生了。如今，程序员在100台机器上编程和在1000台上面编程没有太多的区别，不需要考虑容错性、并发的问题。
3.数据这么多，要如何存储、存储那些数据以及存储多久成为一个难题。通常我们认为，从一个字节上可以获取多少价值，存储它又要花费多少费用，如果两者的比值大于1，就值得存储更多的数据。从大数据的角度来看，存储单条数据的价值可能不大，拥有更多的数据时，数据就有了价值。举一个简单的例子，如果我们只有一个用户的访问日志，显然没有太多价值，但是如果拥有了全站所有用户的访问日志，我们就可以对数据进行分析，从中发现潜在的规律和趋势。
4.如何进行数据分析，这是一个复杂的技术。Spark的很多库就是为了解决不同场景下的分析任务而存在的。比如MLlib库就是为了解决各种机器学习问题开发的库。这些问题包括分类问题、回归问题、聚类等。GraphX就是为了分析社交网络等应用开发的库。
5.大数据时代面临的另一个问题就是大数据交易。数据本身已成为一种可以交易的商品。15年国务院印发了《促进大数据发展行动纲要》，纲要提出要建设公共数据资源开放的统一开放平台。
三.谷歌的大数据解决方案
谷歌的搜索引擎是搜索引擎中的领导者，很重要的原因之一就是谷歌在大数据技术方面的领先。2003年谷歌先后发表了GFS、MapReduce、BigTable等几篇论文。其中的MapReduce范式提供了在集群环境中的可扩展实现，使得程序员可以利用集群处理更多的数据，而不需要考虑容灾和并发的问题。谷歌的三大论文引发了人们在大数据领域的大量研究，直接导致了Hadoop的出现。如今的Hadoop，包括MapReduce与分布式文件系统【HDFS】已经成为数据处理的事实标准。大量的工业界应用，例如腾讯、百度、华为、AT&T等都有自己的Hadoop集群。
MapReduce能做的事情很多，包括实现各种机器学习算法，但MapReduce不是唯一的大数据分析范式，有一些场景是不适合使用MapReduce的，比如处理网状的数据结构时，这就要求能够处理顶点和边的增加和减少操作，并在所有节点上进行运算。典型的场景是在搜索引擎中处理地图计算和进行社交网络分析。谷歌也建设了基于图的计算系统Pregel，允许连通的节点之间互相交换信息。Pregel基本运算是节点之间的超级步骤【superstep】，每个顶点都有一个用户自定义的计算函数和值，所有的边都可以并行计算，顶点可以通过边来发送消息并与其它顶点交互数据，可以聚合所有节点的信息，计算最大最小值等。
当然，并不是说谷歌出现之前就不存在大数据。很多我们使用的方法和技术都是过去已有的，比如分布式系统广泛应用的Paxos算法，是莱斯利·兰伯特于1990年提出的一种基于消息传递的一致性算法。
四.Hadoop生态系统
Hadoop是谷歌大数据解决方案的开源实现，使用Java语言开发，其核心主要是分布式文件系统【HDFS】和批处理计算框架【MapReduce】。在此基础上还有一些重要的系统对其进行支撑。
1.Hive
Hive是在HDFS和MapReduce上提供一个类似SQL风格的抽象层，非常容易上手。用户可以用数据库、表的概念来管理数据，使用SQL来访问、计算。不需要写MapReduce程序。SQL语法非常类似于关系型数据库，支撑常见的select、join、group by、insert等操作。
2.HBase
HBase是基于Hadoop的非关系型数据库【基于BigTable思想设计】，具备分布式、可扩展的特点，支撑在几十亿行、数百万列的一张大表上进行实时、随机地读写访问。典型场景有各种数据仓库，比如淘宝用户历史订单等。
3.Zookeeper
Zookeeper是提供分布式应用程序协调服务的系统，是谷歌的Chubby的一个开源实现，是Hadoop和HBase的重要组件。在下面要讲的Spark计算框架也可以使用Zookeeper进行协调控制。
Hadoop是一个批处理系统，不擅长实时计算，如果需要实时或准实时的分析，可以使用Storm【Twitter】、S4【雅虎】、Akka等系统。另外，Hadoop也不擅长复杂数据结构计算，比如图计算，这时可以使用开源系统中的GraphLab或Spark的GraphX库。
从Hadoop2.x开始，Hadoop YARN将资源调度从MapReduce范式中分离出来。YARN已经成为通用的资源管理系统，可以为上层应用提供统一的资源管理和调度，Spark支持部署在YARN管理的集群上。
在工业界大规模应用Hadoop生态的系统，还要面临部署、排错、升级等问题。为解决这些问题，降低Hadoop的使用门槛，我们可以使用Hadoop商用解决方案提供商的产品，目前比较成熟的提供商有Cloudera、Hortonworks和MapR等。
五.Spark的起源
机器学习算法通常需要对同一个数据集进行多次迭代计算，而MapReduce中每次迭代都会涉及HDFS的读写，以及缺乏一个常驻的MapReduce作业，因此每次迭代需要初始化新的MapReduce任务，这就导致MapReduce执行的效率不高。同时，基于MapReduce之上的Hive、Pig等技术也存在这样类似的问题。
因此，Spark项目就应运而生了，Spark作为一个研究项目，诞生于加州大学伯克利分校AMP实验室。2009年Spark论文发布，在某些任务表现上，Spark相对于Hadoop MapReduce有10~20倍的性能提升。2010年Spark开源，且在开源社区下发展迅速。2014年，Spark1.0正式发布，现如今是Apache基金会的顶级项目。
六.Spark的特点
Spark在刚出现时，常常被冠以内存计算，这不是没有缘由的。在典型应用中，Spark读取HDFS中的文件，加载到内存，在内存中使用弹性分布式数据集【Resillient Distributed Dataset RDD】来组织数据。RDD可以重用，支持重复的访问，在机器学习的各个迭代中它都会驻留内存，这样能显著地提升性能。即使是必须使用磁盘进行复杂计算的场景，Spark也常常比Hadoop MapReduce快的多。
例如，在2015年的排序大赛：
Spark一直寻求保持Spark引擎小而紧凑。Spark0.3只有3900行代码，其中1300行为Scala解析器，600行为示例代码，300行为测试代码。即使在今天，Spark核心代码也只有10万行左右，因此更容易为许多开发者所理解和供我们改变和提高。
Spark是一个通用计算框架，包含特定场景下的计算库：Streaming、SQL、MLlib、GraphX等。除了支持常见的MapReduce范式，它还能够支持图计算。流计算等复杂计算场景，在很大程度上弥补了Hadoop的不足。
在容错方面，Spark也有自己的特色。容错机制又称血统【Lineage】容错，即记录创建RDD的一系列变换序列，每个RDD都包含了它是如何由其他RDD变换过来的，而且包括如何重建某一块数据的信息。由于Spark只允许进行粗粒度的RDD转换，所以其容错机制相对高效。Spark也支持检查点【checkpoint】的容错机制。
Spark自带调度器，同时能够运行在Hadoop YARN集群、Mesos等之上，很方便地和现有的集群进行融合。
Spark的输入支持本地存储、Hadoop的HDFS，以及其他支持Hadoop接口的系统：S3、Hive、HBase等。Spark还有一个优点，就是当RDD的大小超出集群的所有内存时，可以优雅地进行降级支持，存储在磁盘。
七.Spark案例代码
1 package big.data.analyse.sparksql 2 3 import org.apache.spark.sql.{Row, SparkSession} 4 import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType} 5 6 /** 7 * Created by zhen on 2018/11/8. 8 */ 9 object SparkInFuncation { 10 def main(args: Array[String]) { 11 val spark = SparkSession.builder().appName("spark内置函数") 12 .master("local[2]").getOrCreate() 13 val userData = Array( 14 "2015,1,www.baidu.com", 15 "2016,4,www.google.com", 16 "2017,3,www.apache.com", 17 "2015,6,www.spark.com", 18 "2016,2,www.hadoop.com", 19 "2017,8,www.solr.com", 20 "2017,4,www.hive.com" 21 ) 22 val sc = spark.sparkContext 23 val sqlContext = spark.sqlContext 24 val userDataRDD = sc.parallelize(userData) // 转化为RDD 25 val userDataType = userDataRDD.map(line => { 26 val Array(age, id, url) = line.split(",") 27 Row( 28 age, id.toInt, url 29 ) 30 }) 31 val structTypes = StructType(Array( 32 StructField("age", StringType, true), 33 StructField("id", IntegerType, true), 34 StructField("url", StringType, true) 35 )) 36 // RDD转化为DataFrame 37 val userDataFrame = sqlContext.createDataFrame(userDataType,structTypes) 38 39 import org.apache.spark.sql.functions._ 40 userDataFrame 41 .groupBy("age") // 分组 42 .agg(("id","sum")) // 求和 43 .orderBy(desc("age")) // 排序 44 .show() 45 } 46 }
八.执行结果
