程序员的自我修养
Home » Apache Spark » Spark Streaming小结

Spark Streaming小结

8条评论34,566次浏览

概述

Spark Streaming类似于Apache Storm,用于流式数据的处理。根据其官方文档介绍,Spark Streaming有高吞吐量和容错能力强这两个特点。Spark Streaming支持的数据输入源很多,例如:Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等等。数据输入后可以用Spark的高度抽象原语如:map、reduce、join、window等进行运算。而结果也能保存在很多地方,如HDFS,数据库等。另外Spark Streaming也能和MLlib(机器学习)以及Graphx完美融合。
streaming-arch

其内部工作方式如下:streaming-flow

Word Count示例

启动Netcat

启动Spark Streaming Application

若在本地调试,可在IDE中启动,否则,用如下命令启动:

测试

输入:

结果:

DStream

Discretized Stream是Spark Streaming的基础抽象,代表持续性的数据流和经过各种Spark原语操作后的结果数据流。在内部实现上,DStream由连续的序列化RDD来表示。每个RDD含有一段时间间隔内的数据,如下图:
streaming-dstream

对数据的操作也是按照RDD为单位来进行的,如下图所示:
streaming-dstream-ops

上图下方的RDD都是通过Spark高级原语的转换而来,计算过程由Spark engine来完成。

Operations

DStream上的原语与RDD的类似,分为Transformations(转换)和Output Operations(输出)两种,此外转换操作中还有一些比较特殊的原语,如:updateStateByKey()、transform()以及各种Window相关的原语。

UpdateStateByKey Operation

UpdateStateByKey原语用于记录历史记录,上文中Word Count示例中就用到了该特性。若不用UpdateStateByKey来更新状态,那么每次数据进来后分析完成后,结果输出后将不在保存。如,若将上文代码示例中的第15行若替换为:

那么输入:hellow world,结果则为:(hello,1)(world,1),然后输入hello spark,结果则为(hello,1)(spark,1)。也就是不会保留上一次数据处理的结果。

使用UpdateStateByKey原语需要用于记录的State,可以为任意类型,如上例中即为Optional<Intege>类型;此外还需要更新State的函数,可参考Word Count示例中的15-20行。

Transform Operations

Transform()原语允许DStream上执行任意的RDD-to-RDD函数。通过该函数可以方便的扩展Spark API。此外,本篇开头所提到的MLlib(机器学习)以及Graphx也是通过本函数来进行结合的。官方示例:

Window Operations

Window Operations有点类似于Storm中的State,可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming的允许状态。如下图所示:streaming-dstream-window

如以下代码表示,每10秒钟处理最近30秒钟中的数据。

Window相关API有:

  • window(windowLength, slideInterval)
  • countByWindow(windowLength, slideInterval)
  • reduceByWindow(func, windowLength, slideInterval)
  • reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])
  • reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
  • countByValueAndWindow(windowLength, slideInterval, [numTasks])

Output Operations

当某个Output Operations原语被调用时,stream才会开始真正的计算过程。现阶段支持的Output方式有以下几种:

  • print()
  • foreachRDD(func)
  • saveAsObjectFiles(prefix, [suffix])
  • saveAsTextFiles(prefix, [suffix])
  • saveAsHadoopFiles(prefix, [suffix])

其它特性

输入源

除了前文中Word Count示例中用到的TCP套接字连接连接作为输入源以外,Spark Streaming还可以使用很多其它的输入源。例如对于文件,可以这样处理:

Spark Streaming将会监控该文件夹,要使用该特性,需要注意以下几点:

  • 该文件夹下的所有文件必须有相同的数据格式
  • 在该文件夹下创建文件的方式必须是原子性的移动或重命名的方式,不可以先创建文件后在进行写入
  • 所有文件夹下的文件不可进行改动

其它数据源的使用可以参考Spark安装包中的examples文件夹中的streaming部分。同样对于特殊的数据输入源,可以进行定制

监控

一般来说,使用Spark自带的Web UI就能满足大部分的监控需求。对于Spark Streaming来说,以下两个度量指标尤为重要(在Batch Processing Statistics标签下):

  • Processing Time:处理每个batch的时间
  • Scheduling Delay:每个batch在队列中等待前一个batch完成处理所等待的时间

若Processing Time的值一直大于Scheduling Delay,或者Scheduling Delay的值持续增长,代表系统已经无法处理这样大的数据输入量了,这时就需要考虑各种优化方法来增强系统的负载。

持久化

与RDD一样,DStream同样也能通过persist()方法将数据流存放在内存中,这样做的好处是遇到需要多次迭代计算的程序时,速度优势十分的明显。而对于上文中提到的各种window原语,其默认的持久化策略就是保存在内存中。

当数据源来自于网络时(例如通过Kafka、Flume、sockets等等),默认的持久化策略是将数据保存在两台机器上,这也是为了容错性而设计的。

参考资料

(转载本站文章请注明作者和出处 程序员的自我修养 – SelfUp.cn ,请勿用于任何商业用途)
分类:Apache Spark
标签:
8条评论
  1. aaa说道:

    :evil:
    这个例子也就只能基于nc的测试而已~~~·

  2. 匿名说道:

    楼主,请问这句代码哪里来:
    StreamingExamples.setStreamingLogLevels();

    请指教

  3. 匿名说道:

    17/02/09 01:15:02 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
    卡在这一行 不动了
    给个高招吧

发表评论


profile
  • 文章总数:79篇
  • 评论总数:331条
  • 分类总数:31个
  • 标签总数:44个
  • 运行时间:1072天

大家好,欢迎来到selfup.cn。

这不是一个只谈技术的博客,这里记录我成长的点点滴滴,coding、riding and everthing!

最新评论
  • 增达网: 受教了!呵呵!
  • Anonymous: :!: :smile: :oops: :grin: :eek: :shock:
  • 27: :razz: dsa会报错,rsa不会
  • Anonymous: 看错了 忽略我
  • Anonymous: UserSideCF这个类在哪里
  • 晴子: 在主节点初始化CM5数据库的时候报错误:Verifying that we can write to /opt/cm-5.9.0/etc/cloudera-scm -server log4j:ERROR Could not...
  • zhangnew: 就4题 :?:
  • linxh: “ 但要是遇到预先并不知道数组的长度而又需要获取正确的(或者称之 为原始的)split长度时,该如何处理呢。。? ” 印象中可以split函数参数传-1?
  • linxh: 班门弄斧一下: ssh host cmd 和直接ssh上后cmd结果不一样是因为ssh直接运行远程命令 是非交互非登录模式与ssh上去得到一个登录交互式Shell二 者加载的环境变量不一样。
  • 匿名: 其实文本分类和数字分类是一样的,只是文本分类需要多一个步骤, 就是计算它的tf-idf值将其转换为double类型
  • yurnom: 可能苹果最近又改变了返回值吧,最近没做测试了。 BadDeviceToken一般测试环境和正式环境弄错的情况 下会出现。
  • Anonymous: :razz: 博主,良心贴啊, 最近也在弄apns推送。 有个问题想请教你一下啊。 你博客中写的 Unregistered 错误,有准确的说明吗, 我看你博客中写的:...
  • 一波清泉: 回复邮箱: 1004161699@qq.com 多谢
  • Anonymous: 17/02/09 01:15:02 WARN Utils: Service ‘SparkUI’ could not bind on port 4040. Attempting port...
  • pacificLee: :twisted:
  • 小码: 为什么没有后面的呢,只有前10个
  • Anonymous: :lol:
  • Anonymous: :razz: 楼主是属于会聊天的。 我想问,sqoop发了几个版本了,应该没这些问题了吧。
  • Anonymous: Config.kafkaConfig.kafkaGroupI d 这个是指自己配置的group id 还是从 import org.apache.kafka.common.config .Config 这个类...
  • Anonymous: ZkUtils.getPartitionsForTopics (zkClient, Config.kafkaConfig.topic) 那个方法是在 spark-streaming_2.10 中 kafka...