Spark介绍

Spark介绍背景介绍起源于美国加州大学伯克利分校AMPLab的大数据计算平台目前是Apache软件基金会的顶级项目Spark是一个分布式计算框架，专注于分布式计算目前版本是1.6.1（2016.04.29）应用企业1、腾讯场景：广告推荐、日志精准查询2、阿里（淘宝）推荐算法、机器学习、graphx场景：基于度分布的中枢节点发现、基于最大连通图的社区发现、基于三角形计数的关系衡量、基于随机游走的用户属性传播等3、雅虎集群有112台节点，9.2TB内存场景：广告推荐（logisticregression回归算法）、shark即席查询4、电商网站场景：网站推荐系统期望能实时根据顾客的点击行为分析其购买意愿，做到精准营销。有空分享下spark在国内各大公司的具体使用情况Whyspark架构先进采用scala编写，底层采用了actormodel的akka作为通讯框架，代码简洁高效基于DAG（有向无环图）的执行引擎，减少了多次计算中间结果写入hdfs的开销建立在统一抽象的RDD之上，使得可以以基本一致的方式应对不同的大数据处理场景执行高效提供Cache机制来支持需要反复迭代的计算，或多次数据共享，减少数据读取的IO开销与Hadoop的MapReduce相比，Spark基于内存的运算，比MR要快100倍；而基于硬盘的运算也要快10倍!WhySpark简洁易用提供了广泛的数据操作类型（20+种），不像MapReduce只有map和reduce两种支持java、scala、pythonapi，支持交互式的scala、pythonshell整体解决方案以RDD强大的表现能力，逐渐形成了一套自己的生态圈，提供了full-statck的解决方案主要包括Spark内存中批处理，SparkSQL交互式查询、SparkStreaming流式计算，GraphX和MLlib提供的常用图计算和机器学习算法与Hadoop无缝衔接Spark可以使用YARN作为它的集群资源管理器读取HDFS、HBase等数据源Spark生态圈spark的数据源支持hdfs、hbase，还包括s3、cassdra、mongoDB、mysql等Spark解决的问题HadoopSpark解决方案批处理MapReduceSpark即席查询HiveShark--SparkSQL流处理实时计算StormSparkStreaming机器学习MahoutMLlib图形计算GraphLabGraphXSpark批处理弹性分布数据集RDDSpark中对数据和计算的抽象，它表示已被分片（partition）、不可变的、并能被并行执行的数据集合每个RDD包含5个主要的特性：a、一组分片（partitions）b、一个计算分片的函数c、依赖（lineage，谱系，怎么来的）d、位置偏好（可选，如何就近拉数据）e、partitioner（可选，对计算结果如何分发）RDDOperation操作作用于RDD上的Operation分为转换(transformation)和动作(action)。spark的所有“转换”都是惰性的，在执行“转换”操作，并不会提交Job，只有在执行“动作”操作，所有的operation才会真正的被执行Operation转换动作从现有的数据集创建一个新的数据集在数据集上运算后，返回一个值给client程序Spark流程示意RDD操作--transformation函数描述transformationmapReturnanewdistributeddatasetformedbypassingeachelementofthesourcethroughafunctionfuncfilterReturnanewdatasetformedbyselectingthoseelementsofthesourceonwhichfuncreturnstrue.flatMapSimilartomap,buteachinputitemcanbemappedto0ormoreoutputitems(sofuncshouldreturnaSeqratherthanasingleitem).sampleSampleafractionfractionofthedata,withorwithoutreplacement,usingagivenrandomnumbergeneratorseed.groupByKeyreduceByKeyunionjoincoGroupsWhencalledondatasetsoftype(K,V)and(K,W),returnsadatasetof(K,(IterableV,IterableW))tuples.ThisoperationisalsocalledgroupWith.partitionByReshufflethedataintheRDDrandomlytocreateeithermoreorfewerpartitionsandbalanceitacrossthem.Thisalwaysshufflesalldataoverthenetwork.cartesianWhencalledondatasetsoftypesTandU,returnsadatasetof(T,U)pairs(allpairsofelements).RDD操作--action函数描述actioncount计算元素的总个数collect转换为ListTreduceAggregatetheelementsofthedatasetusingafunctionfunc(whichtakestwoargumentsandreturnsone).Thefunctionshouldbecommutative（交换律）andassociative（结合律）sothatitcanbecomputedcorrectlyinparallel.结合律(a+b)+c=a+(b+c)交换律a+b=b+asaveforeachsaveAsTextFilecountByKeyfirsttaketakeSample共享变量1）广播变量Broadcastint[]broadcastVar=sc.broadcast(newint[]{1,2,3});System.out.println(broadcastVar.value());注：广播变量应该可以序列化，并且应该是小数据，大数据建议使用rddjoin方式2）累加器AccumulatorIntegeraccum=sc.accumulator(0);sc.parallelize(Arrays.asList(1,2,3,4)).foreach(x-accum.add(x));System.out.println(accum.value());注：一般是Integer或LongRDD持久化是否持久化默认情况下，每一个转换后的rdd，在它之上执行新的动作时被重新计算如果一个rdd被多次使用，则需要考虑持久化；另外，shuffle后生成的rdd也应持久化，因为shuffle代价高持久化效果rdd持久化之后，下次查询rdd时，它将能被快速访问，不需要计算如果持久化无谓的rdd，则会浪费内存或硬盘空间，反而降低性能持久化方法可以调用persist或cache方法persist需指定StorageLevel，主要有：Memory、Memory_ONLY_SER\DISK_ONLY、MEMORY_AND_DISK、MEMORY_AND_DISK_SER、OFF_HEAP(Tachyon)kryo:ListTuple2Integer,Integerlist=newList()1000个元素需要1000*（4+4+8+8+8）+8=40008个字节，但是使用kyro会按列转为数组，只需（1000*4+8）*2+8=8024批处理演示--WordCountJavaDemo批处理演示--WordCountPythonDemoiPython地址为:程序执行本机调试Spark客户端spark-submit--masteryarn-client--classmain.WordCountAppWordCount.jar--num-executors3--driver-memory512m--executor-memory512m--executor-cores1Oozie执行hue的Workflow：执行过程--DAG一个典型的批处理过程如下图，该图被称为DAG(有向无环图)Spark执行过程--DependencySpark执行过程--StageSpark执行过程--容错机制Spark执行过程--TaskSchedule、TaskSet客户端上的SparkContext拥有DAGScheduler的实例，在runJob方法中会进一步调用DAGScheduler的runJob方法。在此时，DAGScheduler会生成DAG和Stage，将Stage提交给TaskScheduler（集群管理）。TaskSchduler将Stage包装成TaskSet，发送到Worker节点进行真正的计算，同时还要监测任务状态，重试失败和长时间无返回的任务。Spark执行过程--TaskSchedule、TaskSetclustermode运行模式运行模式包括：local、Yarn、Mesos、Standalone(Spark自带)目前的使用情况调试：local生产：Yarn一般使用yarn-client，driver在客户端，pythonspark不支持yarnclusterStandalone模式基于Yarn的Spark集群Shuffle（混洗、洗牌）SparkSQLSparkSQL发展--HiveHive，第一个sqlonhadoop工具，给熟悉DBMS但又不理解MapReduce的技术人员提供快速上手的工具缺点内核为MapReduce，因大量的磁盘IO而效率低下SparkSQL发展--Shark1、其他SqlonHadoop工具，表现突出的有：MapR的Drill、Cloudera的Impala、Shark2、Shark是Spark生态中的一个组件，它参考Hive架构，并修改了内存管理、物理计划、执行三个模块，使得性能相较Hive有10-100倍的提升SparkSQL发展历程Shark-SparkSQL因Shark严重依赖Hive，限制了与Spark其他组件的集成，故而停止Shark，提出SparkSQL，从而在以下方面得到提升数据兼容：Hive、RDD、Parquet、json，还支持关系型数据库（jdbc）性能优化：除In-MemoryColumnarStorage、byte-codegeneration等优化技术外、还引进CostModel对查询进行动态评估、获取最佳物理计划等组件扩展：sql的语法解析器、分析器、优化器可以重新定义Hive-HiveonSparkHive将不再受限于一个引擎，可以采用Map-Reduce、Tez、Spark等引擎性能改进：Hive--Shark性能改进:Shark--SparkSQLWhysofast内存列存储（In-MemoryColumnarStorage）字节码生成技术（bytecodegeneration，即CG）Scala代码优化SparkSQL运行架构SparkSQL结构化数据模块，提供DataFrame作为分布式SQL查询引擎DataFrame来源可以是rdd，json、jdbcSparkSQL可以直接读取hivemetastore，从而直接使用sparksqlSparkSQL有自己的clishellSparkThriftServerThriftServer是一个JDBC/ODBC接口，用户可以通过JDBC