spark分享分析48页PPT文档课件.ppt

1、目录Spark简介Spark批处理Spark集群模式SparkSQLSpark StreamingSpark简介Spark是什么Spark特点Spark生态系统Spark与Hadoop的区别PPT模板下载：1ppt/moban/ Spark是什么官网介绍：官网介绍：Apache Spark is a fast and general engine for large-scale data processing.Spark是加州大学伯克利分校AMP实验室开发的通用内存并行计算框架，分布式资源管理工作交由集群管理工具（Mesos、YARN）PPT模板下载：1ppt/moban/ Spark特点1.

2、先进架构：Spark采用Scala语言编写，底层采用actor mode的akka作为通讯架构，代码十分简洁高效。基于DAG图的执行引擎，减少多次计算之间中间结果写到HDFS的开销。建立在统一抽象的RDD(分布式内存抽象)之上，使得它可以以基本一致的方式应对不同的大数据处理场景。2.运行速度快：提供Cache机制来支持需要反复迭代的计算，减少数据读取的IO开销3.易用性好：Spark提供广泛的数据集操作类型(各种转换算子，行动算子等)Spark支持Java，Python和Scala API，支持交互式Python和Scala的shell4.通用性强：以其RDD模型的强大表现能力，逐渐形成了一套

3、自己的生态圈，提供了full-stack的解决方案。主要包括Spark内存中批处理，Spark SQL交互式查询，Spark Streaming流式计算，Mllib机器学习算法，GraphX图计算。5.与Hadoop无缝衔接：Spark可以使用YARN作为它的集群管理器读取HDFS,HBASE的Hadoop的数据PPT模板下载：1ppt/moban/ Spark生态圈也称为BDAS（伯克利数据分析栈），是伯克利APMLab实验室打造的，力图在算法（Algorithms）、机器（Machines）、人（People）之间通过大规模集成来展现大数据应用的一个平台。Spark生态圈以Spark Co

4、re为核心，从HDFS、Cassandra、Amazon S3和HBase等持久层读取数据，以MESS、YARN和自身携带的Standalone为资源管理器调度Job完成Spark应用程序的计算。 这些应用程序可以来自于不同的组件，如Spark Shell/Spark Submit的批处理、Spark Streaming的实时流处理应用、Spark SQL的即席查询、MLlib的机器学习、GraphX的图处理和SparkR的数学计算等等。 Spark生态系统PPT模板下载：1ppt/moban/ Spark与Hadoop的区别HadoopSpark抽象层次低，需要手工编写代码来完成，使用上难以

5、上手基于RDD的抽象，使数据处理逻辑的代码非常简短只提供两个操作，Map和Reduce，表达力欠缺提供很多转换和动作，很多基本操作如Join，GroupBy已经在RDD转换和动作中实现中间结果也放在HDFS文件系统中中间结果放在内存中，内存放不下了会写入本地磁盘ReduceTask需要等待所有MapTask都完成后才可以开始分区相同的转换构成流水线放在一个Task中运行，分区不同的转换需要Shuffle，被划分到不同的Stage中，需要等待前面的Stage完成后才可以开始时延高，只适用Batch数据处理，对于交互式数据处理，实时数据处理的支持不够通过将流拆成小的batch提供Discretiz

6、ed Stream处理流数据对于迭代式数据处理性能比较差通过在内存中缓存数据，提高迭代式计算的性能历经10年发展，已在生产环境稳定运行多年运行不够稳定PPT模板下载：1ppt/moban/ Spark与Hadoop的区别Hadoop数据抽取运算模型：反复读写，磁盘IO是瓶颈Spark数据抽取运算模型：Spark批处理RDD简介Spark程序入口创建RDDRDD操作TransformationsActionsPPT模板下载：1ppt/moban/ RDD (Resilient Distributed Dataset) ：弹性分布式数据集-分布式内存抽象的概念RDD是Spark对数据的核心抽象，是

7、Spark的基石。RDD是一个可容错、只读的、已被分区的、可并行操作的分布式元素集合RDD的特点：1.只读：状态不可变，不能修改2.分区：支持元素根据 Key 来分区 ( Partitioning ) ，保存到多个结点上，还原时只会重新计算丢失分区的数据，而不会影响整个系统3. RDD必须是可序列化的4.路径：在 RDD 中叫血统 ( lineage ) ，即 RDD 有充足的信息关于它是如何从其他 RDD 产生而来的5.持久化：可以控制存储级别（内存、磁盘等）来进行持久化在RDD的内部实现中每个RDD都可以使用5个方面的特性来表示，即数据分区的集合，能根据本地性快速访问到数据的偏好位置，依赖

8、关系，计算方法，是否是哈希/范围分区的元数据：RDD简介操作操作含义含义partitions()返回一组Partition对象preferredLocations(p)根据数据存放的位置，返回分区p在哪些节点访问更快dependencies()返回一组依赖iterator(p, parentIters)按照父分区的迭代器，逐个计算分区p的元素(计算函数)partitioner()返回RDD是否hash/range分区的元数据信息PPT模板下载：1ppt/moban/ SparkContext:Spark应用程序需要做的第一件事就是创建一个 SparkContext 对象，SparkContex

9、t对象决定了Spark如何访问集群。而要新建一个SparkContext对象，你还得需要构造一个 SparkConf 对象，SparkConf对象包含了你的应用程序的配置信息。每个JVM进程中，只能有一个活跃（active）的SparkContext对象。如果你非要再新建一个，那首先必须将之前那个活跃的SparkContext 对象stop()掉。Spark中已经有一个创建好的SparkContext，简称scScala:import org.apache.spark.SparkContext import org.apache.spark.SparkConfval conf=new Spar

10、kConf().setAppName(appName).setMaster(master)val sc = new SparkContext(conf)Python:from pyspark import SparkConf,SparkContextconf=SparkConf().setAppName(appName).setMaster(master)sc=SparkContext(conf=conf)Spark程序入口PPT模板下载：1ppt/moban/ 一、并行化集合一、并行化集合并行化集合是以一个已有的集合对象(例如：Scala Seq)为参数,调用 SparkContext.pa

11、rallelize() 方法创建得到的RDD。集合对象中所有的元素都将被复制到一个可并行操作的分布式数据集中。Scala:Python:val data=Array(1,2,3,4,5)data=1,2,3,4,5val distData=sc.parallelize(data)distData=sc.parallelize(data)二、外部数据集二、外部数据集Spark 可以通过Hadoop所支持的任何数据源来创建分布式数据集，包括：本地文件系统、HDFS、Cassandra、HBase、Amazon S3 等。Spark 支持的文件格式包括：文本文件（text files）、Sequen

12、ceFiles，以及其他 Hadoop 支持的输入格式（InputFormat）。Scala: val distFile = sc.textFile(data.txt)Python: distFile = sc.textFile(data.txt)三、通过转换现有的三、通过转换现有的RDD得到得到四、改变现有四、改变现有RDD的持久性的持久性(cache、save)创建RDDPPT模板下载：1ppt/moban/ 日志挖掘:val lines=sc.textFile(“hdfs:/.”)val errors=lines.filter(_.startsWith(“ERROR”)errors.ca

13、che()errors.filter(_.contains(“HDFS”).map(_.split(t)(3).take(10)作用于RDD上的操作分为转换(transformation)和动作(action)。Spark中的所有transformation都是惰性的，在执行transformation，并不会提交Job，只有在执行action操作，才会被提交到集群中真正的被执行。Transformation：将已有RDD转换得到一个新的RDD 。Action：计算，返回结果或把RDD数据写到存储系统中。RDD操作PPT模板下载：1ppt/moban/ 窄依赖（narrow dependenc

14、ies）：父RDD的每个分区都只被子RDD的一个分区所依赖比如map、filter、union等宽依赖（wide dependencies）：父RDD的分区被多个子RDD的分区所依赖。比如groupByKey、reduceByKey等1.程序优化：窄依赖支持在一个节点上管道化执行。如在filter之后执行map2.容错：窄依赖支持更高效的故障还原。只有丢失的父RDD的分区需要重新计算宽依赖需要所有父RDD的分区，因此就需要完全重新执行Checkpoint：Lineage链较长、宽依赖的RDD需要采用检查点机制。RDD依赖PPT模板下载：1ppt/moban/ Transformation作用作

15、用示例示例结果结果map(func)返回一个新的数据集，其中每个元素都是由源RDD中一个元素经func转换得到的Rdd.map(x=x+1)1,2,3,32,3,4,4filter(func)返回一个新的数据集，其中包含的元素来自源RDD中元素经func过滤后的结果rdd.filter(x=x!=1)1,2,3,32,3,3flatMap(func)类似于map，但每个输入元素可以映射到0到n个输出元素Rdd.flatMap(x=x.split(“ “)“hello world”,”hi”“hello”,”world”,”hi”union(otherDataset)返回源数据集和参数数据集的并

16、集Rdd.union(other)1,2,3 3,4,51,2,3,3,4,5distinct(numTasks)返回对源数据集做元素去重后的新数据集Rdd.distinct()1,2,3,31,2,3groupByKey(numTasks)若源RDD包含 (K, V) 对，则返回一个新的数据集包含 (K, Iterable) 对Rdd.groupByKey()Rdd=(1,2),(3,4),(3,6)(1,2),(3,4,6)reduceByKey(func, numTasks)若源RDD为 (K, V) 对，则为(K, V) 对的RDD，每个key对应的value是经过func聚合后的结果

17、Rdd.reduceByKey(x,y)=x+y)Rdd=(1,2),(3,4),(3,6)(1,2),(3,10)join(otherDataset, numTasks)若源RDD为 (K, V) 且参数RDD为(K, W)，则返回的新RDD中将包含内关联后key对应的 (K, (V, W) 对Rdd.join(other)Rdd=(1,2),(3,4),(3,6)Other=(3,9)(3,(4,9),(3,(6,9)TransformationsPPT模板下载：1ppt/moban/ Action作用作用示例示例1,2,3,3结果结果reduce(func)将RDD中元素按func进行聚

18、合Rdd.reduce(x,y)=x+y)9collect()将数据集中所有元素以数组形式返回驱动器（driver）程序。Rdd.collect()1,2,3,3count()返回数据集中元素个数Rdd.count()4first()返回数据集中首个元素（类似于 take(1) ）Rdd.first()1take(n)返回数据集中前 n个元素Rdd.take(2)1,2saveAsTextFile(path)将数据集中元素保存到指定目录下的文本文件中，支持本地文件系统、HDFS 或者其他任何Hadoop支持的文件系统。saveAsSequenceFile(path)将数据集中元素保存到指定目录

19、下的Hadoop Sequence文件中，支持本地文件系统、HDFS 或者其他任何Hadoop支持的文件系统。saveAsObjectFile(path)将RDD元素以Java序列化的格式保存成文件，保存结果文件可以使用 SparkContext.objectFile 来读取。ActionsSpark集群模式Spark完整示例集群模式概览术语解释Stage划分Spark任务调度Spark运行模式PPT模板下载：1ppt/moban/ Spark完整示例Scala: submit提交：spark-submit -master yarn -class com.xxx AppName.jarimpo

20、rt org.apache.spark.SparkContext import org.apache.spark.SparkConf object SimpleApp def main(args: ArrayString) val logFile = YOUR_SPARK_HOME/README.md val conf = new SparkConf().setAppName(Simple Application) val sc = new SparkContext(conf) val logData = sc.textFile(logFile, 2).cache() val numAs =

21、logData.filter(line = line.contains(a).count() val numBs = logData.filter(line = line.contains(b).count() println(Lines with a: %s, Lines with b: %s.format(numAs, numBs) Python: submit提交：spark-submit -master yarn -executor -memory 10g AppName.pyfrom pyspark import SparkContextlogFile = YOUR_SPARK_HO

22、ME/README.md sc = SparkContext(local, Simple App)logData = sc.textFile(logFile).cache() numAs = logData.filter(lambda s: a in s).count() numBs = logData.filter(lambda s: b in s).count() print(Lines with a: %i, lines with b: %i % (numAs, numBs)PPT模板下载：1ppt/moban/ 在Spark中由SparkContext负责和ClusterManager

23、通信，进行资源的申请、任务的分配和监控等；当Executor部分运行完毕后，Driver负责将SparkContext关闭。Spark会为该应用在各个集群节点上申请executor，用于执行计算任务和存储数据。接下来，Spark将应用程序代码（JAR包或者Python文件）发送给所申请到的executor 。最后SparkContext将分割出的task发送给各个executor去运行。集群模式概览PPT模板下载：1ppt/moban/ 注意：1.每个Spark应用程序都有其对应的多个executor进程， executor进程在整个应用程序生命周期内，都保持运行状态，并以多线程方式运行所收到

24、的任务。好处：可以隔离各个Spark应用，从调度角度来看，每个driver可以独立调度本应用程序内部的任务，从执行器角度来看，不同的Spark应用对应的任务将会在不同的JVM中运行。坏处：多个Spark应用程序之间无法共享数据，除非把数据写到外部存储中。2. Spark对底层的ClusterManager一无所知。只要Spark能申请到executor进程，并且能与之通信即可。3.driver在整个生命周期内必须监听并接受其对应的各个Executor的连接请求。因此，driver必须能够被所有worker节点访问到。4.因为集群上的任务是由driver来调度的，所以driver应该和worke

25、r节点距离近一些，最好在同一个本地局域网中。如果你需要远程对集群发起请求，最好还是在driver节点上启动RPC服务，来响应这些远程请求，同时把driver本身放在集群worker节点比较近的机器上。集群模式概览PPT模板下载：1ppt/moban/ 术语术语描述描述Application用户编写的Spark应用程序，包含一个Driver program和分布在集群中多个节点上运行的若干ExecutorDriver program运行Application的main()函数并且创建SparkContext Cluster manager在集群上获取资源的外部服务(例如：Standalone、M

26、esos、Yarn)Worker node集群中任何可以运行Application代码的节点ExecutorApplication运行在Worker 节点上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上，每个Application都有各自独立的一批ExecutorTask被送到某个Executor上的工作单元Job包含多个Task组成的并行计算，往往由Spark Action催生Stage每个Job会被拆分很多组Task，每组任务被称为Stage，也可称TaskSet，一个作业分为多个阶段DAGScheduler实现将Spark作业分解成一到多个Stage，每个Stage

27、根据RDD的Partition个数决定Task的个数，然后生成相应的Taskset放到TaskScheduler中TaskScheduler与DAGScheduler交互，实现Task分配到Executor上执行术语解释PPT模板下载：1ppt/moban/ Stage划分stage的边界有两种情况：1.宽依赖上的Shuffle操作；2.已缓存分区，它可以缩短父RDD的计算过程。一个stage的开始就是从外部存储或shuffle结果中读取数据，一个stage的结束就是发生shuffle或生成结果时PPT模板下载：1ppt/moban/ 1.创建RDD，经过一系列Transformation，最

28、后Action2.Action会触发SparkContext的rujob方法，交给DAGScheduler处理3.DAGScheduler将DAG划分成Stage4.将Stage交给TaskScheduler5.集群的Executor上运行Spark任务调度PPT模板下载：1ppt/moban/ 一个按 A-Z 首字母分类，查找相同首字母下不同姓名总个数的例子：步骤 1 ：创建 RDD 上面的例子除去最后一个 collect 是个动作，不会创建 RDD 之外，前面四个转换都会创建出新的 RDD 。因此第一步就是创建好所有 RDD( 内部的五项信息 ) 。 步骤 2 ：创建执行计划 Spark

29、会尽可能地管道化，并基于是否要重新组织数据来划分 阶段 (stage) ，例如本例中的 groupByKey() 转换就会将整个执行计划划分成两阶段执行。最终会产生一个 作为逻辑执行计划。步骤 3 ：调度任务将各阶段划分成不同的 任务 (task) ，每个任务都是数据和计算的合体。在进行下一阶段前，当前阶段的所有任务都要执行完成。sc.textFile(“hdfs:/names”) .map(name=(name.charAt(0),name) .groupByKey() .mapValues(name=names.toSet.size) .collect()RDD执行过程PPT模板下载：1p

30、pt/moban/ 运行环境运行环境模式模式描述描述Local单机模式常用于本地开发测试，分为local单线程和local-cluster多线程Standalone集群模式Spark自带，最简单的集群模式Hadoop Yarn集群模式运行在Yarn资源管理器之上，由Yarn负责资源管理，Spark负责任务调度和计算Apache Mesos集群模式运行在Mesos资源管理器之上，由Yarn负责资源管理，Spark负责任务调度和计算Amazon Ec2集群模式运行在云端的集群Spark运行模式Spark SQLSparkSQL简介SparkSQL入口DataFrame简介DataFrame创建Da

31、taFrame操作和RDD互操作PPT模板下载：1ppt/moban/ SQL-on-Hadoop：Hive是SQL-on-Hadoop最常用的工具，但是MapReduce计算过程中大量的中间磁盘落地过程消耗了大量的I/O，降低的运行效率，为了提高SQL-on-Hadoop的效率，大量的SQL-on-Hadoop工具开始产生，其中表现较为突出的是：MapR的DrillCloudera的ImpalaHortonworks的Hive on TezFacebook的PrestoSparkSQL(Shark-Hive on Spark)但Shark对于Hive的太多依赖，制约了Spark各个组件的相互

32、集成，所以提出了SparkSQL项目。SparkSQL简介PPT模板下载：1ppt/moban/ 入口：入口：SQLContext:SparkSQL的所有功能入口都是SQLContext类及其子类。要创建一个SQLContext对象，首先需要有一个SparkContext对象。Scala:Val sc:SparkContext /假设已经有一个SparkContext对象Val sqlContext = new org.apache.spark.sql.SQLContext(sc)Python:sqlContext=SQLContext(sc)HiveContext:HiveContext继承

33、自SQLContext，除了SQLContext的功能之外， HiveContext还提供了完整的HiveQL语法，UDF使用，以及对Hive表中数据的访问。Scala:Val HiveContext = new org.apache.spark.sql. hive.HiveContext(sc)Python:HiveContext=HiveContext(sc)SparkSQL入口PPT模板下载：1ppt/moban/ DataFrame：是一种分布式数据集合，每一条数据都由几个命名字段组成。概念上来说，她和关系型数据库的表 或者 R和Python中的data frame等价，只不过在底层，

34、DataFrame采用了更多优化。DataFrame可以从很多数据源加载数据并构造得到，如：结构化数据文件，Hive中的表，外部数据库，或者已有的RDD。相对于RDD，DataFrame有几个特点：1.包含schema信息，能够进行针对性的优化。 2.对用户有更加友好、更直观的API。3.与外部数据源 API紧密集成，可以用作多种存储格式和存储系统间的数据交换媒介。DataFrame简介PPT模板下载：1ppt/moban/ 创建DataFrame1.从json文件创建DataFrame：val df = hiveContext.read.json(examples/src/main/reso

35、urces/people.json)2.从parquet文件创建DataFrame：val df = hiveContext.read.parquet(examples/src/main/resources/people.parquet)3.从orc文件创建DataFrame：val df = hiveContext.read.orc(examples/src/main/resources/people.orc)4.从hive表创建DataFrame：val df =hiveContext.table(gdpi)5.从txt文件创建DataFrame：val df =hiveContext.r

36、ead.text(/path/to/spark/README.md)PPT模板下载：1ppt/moban/ DataFrame操作1.展示 DataFrame 的内容 df.show() / age name / null Michael / 30 Andy / 19 Justin2.展示所有人，但所有人的 age 都加1 df.select(df(name), df(age) + 1).show() / name (age + 1) / Michael null / Andy 31 / Justin 20 3.计算各个年龄的人数 df.groupBy(age).count().show()

37、/ age count / null 1 / 19 1 / 30 1PPT模板下载：1ppt/moban/ DataFrame操作SQL操作：1.首先把DataFrame注册为临时表df.registerTempTable(people) 2. HiveContext.sql执行SQL查询，并返回DataFrame,语法与Hql一致val teenagers = hiveContext.sql(SELECT name, age FROM people WHERE age = 13 AND age Person(p(0), p(1).trim.toInt).toDF() people.regis

38、terTempTable(people) / sqlContext.sql方法可以直接执行SQL语句 val teenagers = sqlContext.sql(SELECT name, age FROM people WHERE age = 13 AND age StructField(fieldName, StringType, true)/ 将RDDpeople的各个记录转换为Rows，即：得到一个包含Row对象的RDDval rowRDD = people.map(_.split(,).map(p = Row(p(0), p(1).trim)/ 将schema应用到包含Row对象的R

39、DD上，得到一个DataFrame val peopleDataFrame = sqlContext.createDataFrame(rowRDD, schema)/ 将DataFrame注册为table peopleDataFrame.registerTempTable(people)/ 执行SQL语句 val results = sqlContext.sql(SELECT name FROM people)PPT模板下载：1ppt/moban/ 和RDD互操作编程方式定义编程方式定义Schema:Python:From pyspark.sql import SQLContextFrom p

40、yspar.sql.types import *sqlContext=SQLContext(sc)/加载文件Lines=sc.textFile(“examples/src/main/resources/people.txt”)Parts=lines.map(lambda l: l.split(“,”)/转换每一行为元组People=parts.map(lambda p:(p0,p1.strip()/定义schemaschemaString=“name age”Fields=StructField(field_name,StringType(),True) for field_name in s

41、chemaString.split()Schema=StructType(fields)/将schema应用到RDD上schemaPeople=sqlContext.createDataFrame(people,schema)/注册为表schemaPeople.registerTempTable(“people”)Results=sqlContext.sql(“SELECT name FROM people”)Spark StreamingSpark Streaming简介Spark Streaming与Strom的对比Spark Streaming工作原理Spark Streaming编程离

42、散数据流转换和输出算子PPT模板下载：1ppt/moban/ Spark Streaming简介Spark Streaming能够实现对实时数据流的流式处理，并具有很好的可扩展性、高吞吐量和容错性。支持从多种数据源获取数据，包括Kafk、Flume、Twitter、ZeroMQ、Kinesis 以及TCP sockets，从数据源获取数据之后，可以使用诸如map、reduce、join和window等高级函数进行复杂算法的处理。最后还可以将处理结果存储到文件系统，数据库数据库和现场仪表盘。在“One Stack rule them all”的基础上，还可以使用Spark的其他子框架，如机器学习

43、、图计算等，对流数据进行处理。PPT模板下载：1ppt/moban/ Spark Streaming与Strom的区别StromSpark Streaming传入一个处理一个，吞吐量相对小批处理，吞吐量大实现真正流式实时的处理数据，延迟在秒级以下，实时性很高本质还是批量处理，在短的时间窗口内进行数据实时处理，延迟在秒级左右，实时性相对较弱每个单独的记录当它通过系统时必须被跟踪，所以Storm能够至少保证每个记录将被处理一次，但是在从错误中恢复过来时候允许出现重复记录能够保证每个批处理的所有数据只处理一次，保证数据不会在恢复的时候错乱(批处理重新执行)Clojure语言开发Scala语言开发提供

44、java API提供java、python API2019年开始就在Twitter内部生产环境中使用，已经非常成熟2019年才陆续有一些公司开始试用，但最近越来越多公司已经在生产中使用PPT模板下载：1ppt/moban/ Spark Streaming工作原理Spark Streaming从实时数据流接入数据，再将其划分为一个个小批量供后续Spark engine处理，最终得到处理后的一批批结果数据。 处理流程：1.Spark Streaming把实时输入数据流以时间片t(如1秒)为单位切分成块2. Spark Streaming会把每块数据作为一个RDD，并使用RDD操作处理每一小块数据3

45、.每个块都会生成一个Spark Job处理4.最终结果也返回多块PPT模板下载：1ppt/moban/ Spark Streaming编程入口：StreamingContext 是Spark Streaming的入口。val conf = new SparkConf().setMaster(local2).setAppName(NetworkWordCount) val ssc = new StreamingContext(conf, Seconds(1)/已有spark context的情况val ssc = new StreamingContext(sc, Seconds(1)Python

46、:Sc=SparkContext(“local2”,”NetworkWordCount”)Ssc=StreamingContext(sc,1)步骤：1.创建DStream对象，并定义好输入数据源。2.基于数据源DStream定义好计算逻辑和输出。3.调用streamingContext.start() 启动接收并处理数据。4.调用streamingContext.awaitTermination() 等待流式处理结束。5.你可以主动调用 streamingContextssc.stop() 来手动停止处理流程。注意点：1.一旦streamingContext启动，就不能再对其计算逻辑进行添加或

47、修改。2.一旦streamingContext被stop掉，就不能restart。3.单个JVM虚机同一时间只能包含一个active的StreamingContext。PPT模板下载：1ppt/moban/ Spark Streaming编程Spark-shell中：import org.apache.spark._import org.apache.spark.streaming._val ssc = new StreamingContext(sc, Seconds(1) / 创建StreamingContext，批次间隔为1秒val lines = ssc.socketTextStream

48、(“192.168.5.2, 9998) / 创建一个连接到hostname:port的Dstreamval words = lines.flatMap(_.split(“ ”) / 将每一行分割成多个单词val pairs = words.map(word = (word, 1) / 对每一批次中的单词进行计数val wordCounts = pairs.reduceByKey(_ + _) wordCounts.print() / 将该DStream产生的RDD的头十个元素打印到控制台上ssc.start() / 启动流式计算ssc. awaitTermination() / 等待直到计算

49、终止Python：From pyspark import SparkComtextFrom pyspark.streaming import StreamingContextWords=lines.flatMap(lambda line:line.split(“ ”) / 将每一行分割成多个单词Pairs=words.map(lambda word: (word,1)/ 对每一批次中的单词进行计数wordCounts=pairs.reduceByKey(lambda x,y: x+y)wordCount.pprint() / 将该DStream产生的RDD的头十个元素打印到控制台上PPT模板下

50、载：1ppt/moban/ 离散数据流离散数据流 (DStreams)Dstream：离散数据流是Spark Streaming最基本的抽象。它代表了一种连续的数据流，要么从某种数据源提取数据，要么从其他数据流映射转换而来。DStream内部是由一系列连续的RDD组成的，每个RDD都是不可变、分布式的数据集。每个RDD都包含了特定时间间隔内的一批数据。任何作用于DStream的算子，其实都会被转化为对其内部RDD的操作。例如，在前面的例子中，我们将 lines 这个DStream转成words DStream对象，其实作用于lines上的flatMap算子，会施加于lines中的每个RDD上，