Spark大数据平台应用框架

胡德禄2016/05/06自我介绍•胡德禄先生，拥有5年以上企业级数据处理及互联网大数据处理的工作经验；•熟悉Spark、Hadoop、Hive、Scala、Python等分布式计算框架及开发工具，处理过上亿用户规模的用户电商数据，10TB级存储规模的数据量；•曾负责并参与灵狐科技的大数据处理平台——保罗（Paul）的设计、研发及优化，访问地址：；•同时也负责参与公司的电商大数据分析平台——ECDT的设计、研发及优化，访问地址：；•这些平台的推出，不仅降低了公司内部使用数据挖掘或机器学习的门槛，同时也提高了工作效率，充分利用大数据技术服务于公司的电商部门及数字营销部门等，同时也服务于合作的公司，促使进一步深入合作；大纲•Spark导论及大数据生态•SparkvsHadoop的计算方式•Spark平台框架及应用场景•Scala语言特性及常用方法•机器学习在Spark平台的应用•Spark和Scala开发示例演示什么是大数据BIGDATA？这些年听过的BIGDATA名词•Hadoop、HDFS、MapReduce•Hbase、Cassandra、MongoDB、Zookeeper•Hive、Pig、Impala、Drill、Mahout•Spark、Tez、Storm、Flink•……SPARK导论及大数据生态HADOOP动物园DISTRIBUTEDCOMPUTEFRAMEWORKS:MAPREDUCE大数据技术框架发展历程THEDATASCIENCEVENNDIAGRAMDEVELOPER20162016:在米国挣钱多的编程语言2016:在数学领域挣钱多的编程语言大纲•Spark导论及大数据生态•SparkvsHadoop的计算方式•Spark平台框架及应用场景•Scala语言特性及常用方法•机器学习在Spark平台的应用•Spark和Scala开发示例演示HADOOPHADOOPSTRENGTHS•LowestCapExsystemforBigData.•Excellentforingestingandintegratingdiversedatasets.•Flexible:fromclassicanalytics(aggregationsanddatawarehousing)tomachinelearning.HADOOPWEAKNESSES•Complexadministration.•YARNcan’tmanagealldistributedservices.•MapReduce:•Haspoorperformance.•Adifficultprogrammingmodel.•Doesn’tsupportstreamprocessing.2013:HADOOPEMBRACESPARK速度！速度！速度！SPARK:MAJORPERFORMANCEIMPROVEMENTSSPARKVSHADOOP•Hadoop:•Spark：兄弟！兄弟！兄弟！MODULESONEOFTHEFASTESTGROWINGOSPROJECTS2015:AGREATYEARFORSPARK后HADOOP时代后HADOOP时代SPARKASTHESTANDARDPROCESSINGENGINEAWINNINGCOMBINATION大纲•Spark导论及大数据生态•SparkvsHadoop的计算方式•Spark平台框架及应用场景•Scala语言特性及常用方法•机器学习在Spark平台的应用•Spark和Scala开发示例演示SPARK平台框架SPARK生态系统SPARK平台框架•SparkCoreEngine：–基于RDD提供丰富的操作接口，利用DAG进行统一的任务规划，使得Spark能够更加灵活处理类似MapReduce的批处理作业。•SparkSQLandDataFrame：–分布式SQL引擎和组织数据。•SparkStreaming：–将流式计算分解为一系列的短小的批处理。•MLlib：–SparkCore天然适合迭代式运算，Mllib就是构建在Spark上的机器学习算法库。•GraphX：–基于Spark的图计算框架，兼容Pregel和GraphLab接口，增强了图构建以及图转换功能。–Spark生态系统的目标就是将批处理、交互式处理、流式处理、图计算等融合到同一个软件栈中。TURNINGBIGDATAINTOVALUE:The‘Datafication’ofourWorld;•Activities•Conversations•Words•Voice•SocialMedia•Browserlogs•Photos•Videos•Sensors•Etc.VolumeVeracityVarietyVelocityAnalysingBigData:•Textanalytics•Sentimentanalysis•Facerecognition•Voiceanalytics•Movementanalytics•Etc.ValueHADOOP应用场景HADOOP应用场景SPARK应用场景•Updateasearchengineinrealtimeaswebpageordocumentschange.•TrainaSPAMfilterwitheveryemail.•Detectanomaliesastheyhappenthroughprocessingoflogsandmonitoringdata.SPARK应用场景SPARK应用场景RDD：核心理论基础•Resilientdistributeddatasets(RDDs)不可变的，按分区组织的数据对象一个RDD可以通过各种（map,filter,groupBy,Join…)操作转换为另一个RDD源头的RDD的数据来源可以是外部存储，如HDFS，也可以动态生成可以采用各种缓存方式加速处理容错,数据本地性,可扩展性THECORE-RESILIENTDISTRIBUTEDDATASETSRDD：核心理论基础建立RDD•将一个集合collection转换为RDD–vala=sc.parallelize(Array(1,2,3))•将一个文本文件转换为RDD–valtestScala=sc.textFile(hdfs:///user/hadoop/testScala.txt)进一步看下RDD•RDDs（可以从一系列数据源创建）•Transformations(延迟执行)•Action(defrunJob[T,U:ClassManifest](rdd:RDD[T],func:Iterator[T]=U):Array[U])TRANSFORMATION和ACTIONRDD：TRANSFORMATIONRDD的TRANSFORMATION•map：所有的pairs都被单独处理valnums=sc.parallelize(Array(1,2,3))valsquares=nums.map(x=x*x)//{1,4,9}•groupby：相同key的分配到同一组，各组之间是单独处理的valschools=sc.parallelize(Seq((sics,1),(kth,1),(sics,2)))schools.groupByKey()//{(sics,(1,2)),(kth,(1))}valsRedByKey=schools.reduceByKey((x,y)=x+y)sRedByKey.foreach(println)//{(sics,3),(kth,1)}RDD：ACTIONRDD的ACTION•返回RDD的所有元素valnums=sc.parallelize(Array(1,2,3))nums.collect()//Array(1,2,3)•返回RDD的前几个元素nums.take(2)//Array(1,2)•返回RDD的元素的数量nums.count()//3•聚合RDD中的元素nums.reduce((x,y)=x+y)//6TRANSFORMATION和ACTION：应用TRANSFORMATION和ACTION：应用TRANSFORMATION和ACTION：应用示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；示例：日志挖掘•将一个日志log的错误信息加载到内存中，并统计相应类型的错误，比如”PHP”、“Mysql”等；SPARKSQL•SparkSql、DataFrames、DataSets；•SQLonSpark；•Catalyst：SparkSQL的查询优化；•SparkSQL示例；SQLONSPARK•SparkSQL：希望成为结构化数据分析的标准。•Shark：已经完成学术使命退出历史舞台。•HiveonSpark：会全面支持现有Hive，只是换了个更加快速的执行引擎--Spark。随着Spark生态系统的发展及壮大，用户会越来越多的采用SQLonSpark作为自身的OLAP解决方案，这也同时表明，未来SQLonSpark的应用和发展会很有想象的空间，尤其是SparkSQL。CATALYST：SPARKSQL的查询优化•Catalyst架构：CATALYST的执行流程SqlPaserSQL语句进行语法解析Analyzer进行属性和关系关联检查Optimizer进行启发式逻辑查询优化QueryPlanner逻辑查询计划转换为物理查询计划prepareForExecution调整数据分布r进入Spark执行空间转换为DAG执行CATALYST的优化：示例•逻辑优化的示例：SELECTClassFROM(SELECTID,Name,ClassFROMSTUDENT)SWHERES.ID=1ProjectClassFilterID=1ProjectID,Name,ClassStudentOptimizationProjectClassFilterID=1StudentSPARKSQL示例：DATAFRAMES（1/3）•将数据集放到Hadoop上：cd$SPARK_HOMEhdfsdfs-put./examples/src/main/resources/user/hadoop/spark•登陆环境：./bin/spark-shell•创建DataFrame：valdf=sqlContext.jsonFile(hdfs:///user/hadoop/spark/resources/people.json)SPARKSQL示例：DATAFRAMES（2/3）•查看DataFrame：df.show()df.printSchema()df.select(name).show()SPARKSQL示例：DATAFRAMES（3/3）•操作DataF