精华内容
参与话题
问答
  • 7.SparkStreaming(上)--SparkStreaming原理介绍.pdf 7.SparkStreaming(下)--SparkStreaming实战.pdf 8.SparkMLlib(上)--机器学习及SparkMLlib简介.pdf 8.SparkMLlib(下)--SparkMLlib实战.pdf 9.SparkGraphX...
  • 前言 引用Spark commiter(gatorsmile)的话:“从Spark-2.X版本后,Spark streaming就进入维护模式,Spark streaming是低阶API,给码农用的,各种坑;Structured streaming是给人设计的API,简单易用。由于太忙,...

    前言

           引用Spark commiter(gatorsmile)的话:“从Spark-2.X版本后,Spark streaming就进入维护模式,Spark streaming是低阶API,给码农用的,各种坑;Structured streaming是给人设计的API,简单易用。由于太忙,所以一直没有在官方文档上 更新说明信息”

            这个坑,我想大部分朋友和我一样-->也已经被坑了很久。既然被坑了,那就看看怎么个坑法,然后探讨一下下他们之间的区别,希望对正在被坑的人有一定的帮助。


    区别

    1、Structured streaming详细介绍

    Databricks的博文:https://databricks.com/blog/2016/07/28/structured-streaming-in-apache-spark.html

    说明文档:https://docs.databricks.com/spark/latest/structured-streaming/index.html

    2、Spark streaming详细介绍

    Databricks的博文:https://databricks.com/blog/2014/09/16/spark-1-1-the-state-of-spark-streaming.html


    3、区别

    参考:

    https://www.quora.com/What-are-the-differences-between-Spark-streaming-and-Spark-structured-streaming-Is-Spark-structured-streaming-the-future-of-Spark-streaming


    结合Kafka使用举例

    1、Spark streaming +Kafka(Spark-2.X以后就没有用例了,推测已经进入维护状态,不再维护)

    Spark-1.6.0版本:http://spark.apache.org/docs/1.6.0/streaming-kafka-integration.html

    2、Structured streaming+Kafka

    Spark-2.1.0版本:http://spark.apache.org/docs/2.1.0/structured-streaming-kafka-integration.html



    展开全文
  • [root@localhost hadoop-1.0.1]# bin/hadoop jar contrib/streaming/hadoop-streaming-1.0.1.jar -file /home/hul/HadoopTestMap -file /home/hul/HadoopTestReduce -mapper /user/root/in/HadoopTestMap -reducer /...
  • Hadoop streaming详解

    千次阅读 2017-03-14 17:20:17
    Hadoop streamingHadoop为MapReduce提供了不同的API,可以方便我们使用不同的编程语言来使用MapReduce框架,而不是只局限于Java。这里要介绍的就是Hadoop streaming API。Hadoop streaming 使用Unix的standard ...

    Hadoop streaming

    Hadoop为MapReduce提供了不同的API,可以方便我们使用不同的编程语言来使用MapReduce框架,而不是只局限于Java。这里要介绍的就是Hadoop streaming API。Hadoop streaming 使用Unix的standard streams作为我们mapreduce程序和MapReduce框架之间的接口。所以你可以用任何语言来编写MapReduce程序,只要该语言可以往standard input/output上进行读写。
    streamming是天然适用于文字处理的(text processing),当然,也仅适用纯文本的处理,对于需要对象和序列化的场景,hadoop streaming无能为力。它力图使我们能够快捷的通过各种脚本语言,快速的处理大量的文本文件。以下是steaming的一些特点:

    • Map函数的输入是通过stand input一行一行的接收数据的。(不像Java API,通过InputFormat类做预处理,使得Map函数的输入是有Key和value的
    • Map函数的output则必须限定为key-value pair,key和value之间用\t分开。(MapReduce框架在处理intermediate的Map输出时,必须做sort和partition,即shuffle
    • Reduce函数的input是Map函数的output也是key-value pair,key和value之间用\t分开。

    常用的Streaming编程语言:

    • bash shell
    • ruby
    • python

    Ruby

    下面是一个Ruby编写的MapReduce程序的示例:

    map

    max_temperature_map.rb:

     #!/usr/bin/env ruby
     STDIN.each_line do |line|
    val = line
    year, temp, q = val[15,4], val[87,5], val[92,1]
    puts "#{year}\t#{temp}" if (temp != "+9999" && q =~ /[01459]/)
    end
    • 从标准输入读入一行data。
    • 处理数据之后,生成一个键值对,用\t分隔,输出到标准输出

    reduce

    max_temperature_reduce.rb:

     #!/usr/bin/env ruby
    last_key, max_val = nil, -1000000
    STDIN.each_line do |line|
    key, val = line.split("\t")
    if last_key && last_key != key
    puts "#{last_key}\t#{max_val}"
    last_key, max_val = key, val.to_i
    else
    last_key, max_val = key, [max_val, val.to_i].max
    end
    end
    puts "#{last_key}\t#{max_val}" if last_key
    • 从标准输入读入一行数据
    • 数据是用\t分隔的键值对
    • 数据是被MapReduce根据key排序之后顺序一行一行读入
    • reduce函数对数据进行处理,并输出,输出仍是用\t分隔的键值对

    运行

    % hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \
    -input input/ncdc/sample.txt \
    -output output \
    -mapper ch02/src/main/ruby/max_temperature_map.rb \
    -reducer ch02/src/main/ruby/max_temperature_reduce.rb
    • hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar指明了使用hadoop streaming
    • hadoop-*-streaming.jar会将input里的文件,一行一行的输出到标准输出。
    • 用-mapper指定Map函数。类似于通过管道将数据传给rb文件: data|ch02/src/main/ruby/max_temperature_map.rb
    • -reducer指定Reduce函数。

    Python

    Map

    #!/usr/bin/env python
    import re
    import sys
    for line in sys.stdin:
    val = line.strip()
    (year, temp, q) = (val[15:19], val[87:92], val[92:93])
    if (temp != "+9999" and re.match("[01459]", q)):
    print "%s\t%s" % (year, temp)

    Reduce

    #!/usr/bin/env python
    import sys
    (last_key, max_val) = (None, -sys.maxint)
    for line in sys.stdin:
    (key, val) = line.strip().split("\t")
    if last_key and last_key != key:
    print "%s\t%s" % (last_key, max_val)
    (last_key, max_val) = (key, int(val))
    else:
    (last_key, max_val) = (key, max(max_val, int(val)))
    if last_key:
    print "%s\t%s" % (last_key, max_val)

    运行

    % hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \
    -input input/ncdc/sample.txt \
    -output output \
    -mapper ch02/src/main/ruby/max_temperature_map.py\
    -reducer ch02/src/main/ruby/max_temperature_reduce.py

    Bash shell

    Map

    #!/usr/bin/env bash
    # NLineInputFormat gives a single line: key is offset, value is S3 URI
    read offset s3file
    # Retrieve file from S3 to local disk
    echo "reporter:status:Retrieving $s3file" >&2
    $HADOOP_INSTALL/bin/hadoop fs -get $s3file .
    # Un-bzip and un-tar the local file
    target=`basename $s3file .tar.bz2`
    mkdir -p $target
    echo "reporter:status:Un-tarring $s3file to $target" >&2
    tar jxf `basename $s3file` -C $target
    # Un-gzip each station file and concat into one file
    echo "reporter:status:Un-gzipping $target" >&2
    for file in $target/*/*
    do
    gunzip -c $file >> $target.all
    echo "reporter:status:Processed $file" >&2
    done
    # Put gzipped version into HDFS
    echo "reporter:status:Gzipping $target and putting in HDFS" >&2
    gzip -c $target.all | $HADOOP_INSTALL/bin/hadoop fs -put - gz/$target.gz

    运行

    % hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \
    -D mapred.reduce.tasks=0 \
    -D mapred.map.tasks.speculative.execution=false \
    -D mapred.task.timeout=12000000 \
    -input ncdc_files.txt \
    -inputformat org.apache.hadoop.mapred.lib.NLineInputFormat \
    -output output \
    -mapper load_ncdc_map.sh \
    -file load_ncdc_map.sh
    
    • 这里的-D mapred.reduce.tasks=0将reduce task观掉,因此也不需要设置-reducer
    • 只使用Mapper,可以通过MapReduce帮助我们并行的完成一些平时只能串行的shell脚本
    • 注意这里的-file,在集群模式下,需要并行运行时,需要-file把文件传输到其他节点

    Combiner

    在streaming模式下,仍然可以运行Combiner,两种方法:

    1. 通过Java编写一个combiner的函数,并使用-combiner option
    2. 以命令行的管道模式完成combiner的任务

    这里具体解释第二种方法:

    % hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \
    -input input/ncdc/all \
    -output output \
    -mapper "ch02/src/main/ruby/max_temperature_map.rb | sort |
    ch02/src/main/ruby/max_temperature_reduce.rb" \
    -reducer ch02/src/main/ruby/max_temperature_reduce.rb \
    -file ch02/src/main/ruby/max_temperature_map.rb \
    -file ch02/src/main/ruby/max_temperature_reduce.rb

    注意看-mapper这一行,通关管道的方式,把mapper的临时输出文件(intermediate file,Map完成后的临时文件)作为输入,送到sort进行排序,然后送到reduce脚本,来完成类似于combiner的工作。这时候的输出才真正的作为shuffle的输入,被分组并在网络上发送到Reduce

    展开全文
  • 前言 又是一个超长的标题(摊手┓( ´∀` )┏)。Spark Streaming 历史比较悠久,也确实非常...反倒是Structured Streaming, 吐槽点比较多,但是到目前,我们经过一番实践,觉得是时候丢掉Spark Streaming 升级到...
        

    前言

    又是一个超长的标题(摊手┓( ´∀` )┏)。Spark Streaming 历史比较悠久,也确实非常好用,更重要的是,大家已经用熟了,有的还做了不少工具了,所以觉得这东西特别好了,不会像一开始各种吐槽了。反倒是Structured Streaming, 吐槽点比较多,但是到目前,我们经过一番实践,觉得是时候丢掉Spark Streaming 升级到Structured Streaming了。

    更新问题

    你看,DB公司已经没怎么对Spark Streaming做更新了。

    API统一

    DStream 和 RDD 看起来很相似,但其实是两个不同的东西,DStream是对RDD在流式计算的里的Wrap。所以流式和批处理,你其实比较难复用的。但是在Structured Streaming中,都是对Dataframe的操作,复杂逻辑处理会很容易的在批处理和流式计算中复用。

    支持实时流式

    Structured Streaming 已经在2.3.0中支持实时流式,潜力可见一斑了。一行代码就可以让原来的微批流转化为实时流。

    同一实例多流支持

    以前我一直希望启动一个spark streaming程序,然后可以动态添加或者删减流,但是在Spark Streaming中,API层次就不允许你这么做。你需要自己重新去封装一套,并且适当的对Kafka那侧做些调整才能达到诉求。而在Structured Streaming中,天生就是多流的管理的。你可以随时停止一个流,启动一个新流,通过API获取流的状态,所有这些,都让流成为Service 变得很容易。StreamingPro实现了流式服务,你可以提交新的流,管理已有的流,参考着mlsql-stream

    更好的限制

    Structured Streaming 是面向Dataframe(表)的,合适的限制会让代码更易于阅读,并且保持更好的运作效率。今天,我们发现,table,sql都是大数据里不可或缺的概念,Structured Streaming 则是更倾向这些概念,而Spark Streaming还是一个面向RDD的东西。

    更好的元数据管理

    我想大家都有自己的offset管理(在Spark Streaming)里,大家的做法五花八门,缺乏标准,Spark Streaming的实现则是一种脑残式实现。在Structured Streaming,这个问题得到了更好的解决。

    对流站在一个更高的抽象层次上

    Spark Streaming一切都在于你自己的代码,而Structured Streaming则为你做了更好的抽象。比如如果结果集不大,那么用complete模式可以保证在一些常见存储中全量覆盖写而实现exactly-once。而wartermark等概念则更是流式计算中常见的诉求。Structured Streaming是站在对流站在一个更好的抽象层次上让你使用的,enjoy它吧。

    一些实践问题

    比如这个Structured Streaming如何实现Parquet存储目录按时间分区,还有就是监控,可能不能复用以前Spark Streaming那套机制了。

    结束语

    是时候丢掉Spark Streaming 升级到Structured Streaming了,让我们享受DB更好的服务。

    展开全文
  • Spark Streaming 转向 Structured Streaming

    千次阅读 2019-05-23 20:33:24
    Spark 团队对 Spark Streaming 的维护将会越来越少,Spark 2.4 版本的 Release Note 里面甚至一个 Spark Streaming 相关的 ticket 都没有。相比之下,Structured Streaming 有将近十个 ticket 说明。所以,是时候...

    导读

    Spark 团队对 Spark Streaming 的维护将会越来越少,Spark 2.4 版本的 Release Note 里面甚至一个 Spark Streaming 相关的 ticket 都没有。相比之下,Structured Streaming 有将近十个 ticket 说明。所以,是时候舍弃 Spark Streaming 转向 Structured Streaming 了,当然理由并不止于此。

    今天这篇文章将重点分析 Spark Streaming 的不足,以及 Structured Streaming 的设计初衷和思想。文章主要参考 2018 年 Sigmod 上面的论文:《Structured Streaming: A Declarative API for Real-Time Applications in Apache Spark》。

    首先可以注意到论文标题中的 Declarative API,中文一般叫做声明式编程 API。一般直接看到这个词可能不知道什么意思,但是当我们列出它的对立单词:Imperative API,中文一般叫命令式编程 API,仿佛一切都明了了。是的,没错,Declarative 只是表达出我们想要什么,而 Imperative 则是说为了得到什么我们需要做哪些东西一个个说明。举个例子,我们要一个糕点,去糕点店直接定做,告诉店员我们要什么样式的糕点,然后店员去给我们做出来,这就是 Declarative。而 Imperative 对应的就是面粉店了。

    0. Spark Streaming 的不足

    在开始正式介绍 Structured Streaming 之前有一个问题还需要说清楚,就是 Spark Streaming 存在哪些不足?总结一下主要有下面几点:

    使用 Processing Time 而不是 Event Time

    首先解释一下,Processing Time 是数据到达 Spark 被处理的时间,而 Event Time 是数据自带的属性,一般表示数据产生于数据源的时间。比如 IoT 中,传感器在 12:00:00 产生一条数据,然后在 12:00:05 数据传送到 Spark,那么 Event Time 就是 12:00:00,而 Processing Time 就是 12:00:05。我们知道 Spark Streaming 是基于 DStream 模型的 micro-batch 模式,简单来说就是将一个微小时间段,比如说 1s 的流数据当成批数据来处理。如果我们要统计某个时间段的一些数据,毫无疑问应该使用 Event Time,但是因为 Spark Streaming 的数据切割是基于 Processing Time,这样就导致使用 Event Time 特别困难。

     Complex Low-level API

    这点比较好理解,DStream (Spark Streaming 的数据模型)提供的 API 类似 RDD 的 API,非常 Low level。当我们编写 Spark Streaming 程序的时候,本质上就是要去构造 RDD 的 DAG 执行图,然后通过 Spark Engine 运行。这样导致的一个问题是,DAG 可能会因为开发者的水平参差不齐而导致执行效率上的天壤之别。这样导致开发者的体验非常不好,也是任何一个基础框架不想看到的(基础框架的口号一般都是:你们专注于自己的业务逻辑就好,其他的交给我)。这也是很多基础系统强调 Declarative 的一个原因。

     Reason about end-to-end application

    这里的 end-to-end 指的是直接 input 到 out,比如 Kafka 接入 Spark Streaming 然后再导出到 HDFS 中。DStream 只能保证自己的一致性语义是 exactly-once 的,而 input 接入 Spark Streaming 和 Spark Streaming 输出到外部存储的语义往往需要用户自己来保证。而这个语义保证写起来也非常有挑战性,比如为了保证 output 的语义是 exactly-once,需要 output 的存储系统具有幂等的特性,或者支持事务性写入,这对于开发者来说都不是一件容易的事情。

     批流代码不统一

    尽管批流本是两套系统,但是这两套系统统一起来确实很有必要,我们有时候确实需要将我们的流处理逻辑运行到批数据上面。关于这一点,最早在 2014 年 Google 提出 Dataflow 计算服务的时候就批判了 streaming/batch 这种叫法,而是提出了 unbounded/bounded data 的说法。DStream 尽管是对 RDD 的封装,但是我们要将 DStream 代码完全转换成 RDD 还是有一点工作量的,更何况现在 Spark 的批处理都用 DataSet/DataFrame API 了。

    1. Structured Streaming 介绍

    Structured Streaming 在 Spark 2.0 版本于 2016 年引入,设计思想参考了很多其他系统的思想,比如区分 processing time 和 event time,使用 relational 执行引擎提高性能等。同时也考虑了和 Spark 其他组件更好的集成。

    Structured Streaming 和其他系统的显著区别主要如下:

    • Incremental query model:Structured Streaming 将会在新增的流式数据上不断执行增量查询,同时代码的写法和批处理 API (基于 Dataframe 和 Dataset API)完全一样,而且这些 API 非常的简单。
    • Support for end-to-end application:Structured Streaming 和内置的 connector 使的 end-to-end 程序写起来非常的简单,而且 "correct by default"。数据源和 sink 满足 "exactly-once" 语义,这样我们就可以在此基础上更好地和外部系统集成。
    • 复用 Spark SQL 执行引擎:我们知道 Spark SQL 执行引擎做了非常多优化工作,比如执行计划优化、codegen、内存管理等。这也是 Structured Streaming 取得高性能和高吞吐的一个原因。

     

     

    2. Structured Streaming 核心设计

    下面我们看一下 Structured Streaming 的核心设计。

    • Input and Output:Structured Streaming 内置了很多 connector 来保证 input 数据源和 output sink 保证 exactly-once 语义。

    而实现 exactly-once 语义的前提是:

    1. Input 数据源必须是可以 replay 的,比如 Kafka,这样节点 crash 的时候就可以重新读取 input 数据。常见的数据源包括 Amazon Kinesis, Apache Kafka 和文件系统。
    2. Output sink 必须要支持写入是幂等的。这个很好理解,如果 output 不支持幂等写入,那么一致性语义就是 at-least-once 了。另外对于某些 sink, Structured Streaming 还提供了原子写入来保证 exactly-once 语义。
    • API:Structured Streaming 代码编写完全复用 Spark SQL 的 batch API,也就是对一个或者多个 stream 或者 table 进行 query。query 的结果是 result table,可以以多种不同的模式(append, update, complete)输出到外部存储中。另外,Structured Streaming 还提供了一些 Streaming 处理特有的 API:Trigger, watermark, stateful operator。
    • Execution:复用 Spark SQL 的执行引擎。Structured Streaming 默认使用类似 Spark Streaming 的 micro-batch 模式,有很多好处,比如动态负载均衡、再扩展、错误恢复以及 straggler (straggler 指的是哪些执行明显慢于其他 task 的 task)重试。除了 micro-batch 模式,Structured Streaming 还提供了基于传统的 long-running operator 的 continuous 处理模式。
    • Operational Features:利用 wal 和状态存储,开发者可以做到集中形式的 Rollback 和错误恢复。还有一些其他 Operational 上的 feature,这里就不细说了。

    3. Structured Streaming 编程模型

    可能是受到 Google Dataflow 的批流统一的思想的影响,Structured Streaming 将流式数据当成一个不断增长的 table,然后使用和批处理同一套 API,都是基于 DataSet/DataFrame 的。如下图所示,通过将流式数据理解成一张不断增长的表,从而就可以像操作批的静态数据一样来操作流数据了。

    在这个模型中,主要存在下面几个组成部分:

    • Input Unbounded Table:流式数据的抽象表示
    • Query:对 input table 的增量式查询
    • Result Table:Query 产生的结果表
    • Output:Result Table 的输出

     

    下面举一个具体的例子,NetworkWordCount,代码如下:

    // Create DataFrame representing the stream of input lines from connection to localhost:9999
    val lines = spark.readStream
    .format("socket")
    .option("host", "localhost")
    .option("port", 9999)
    .load()
    // Split the lines into words
    val words = lines.as[String].flatMap(_.split(" "))
    // Generate running word count
    val wordCounts = words.groupBy("value").count()
    // Start running the query that prints the running counts to the console
    val query = wordCounts.writeStream
    .outputMode("complete")
    .format("console")
    .start()

    代码实际执行流程可以用下图来表示。把流式数据当成一张不断增长的 table,也就是图中的 Unbounded table of all input。然后每秒 trigger 一次,在 trigger 的时候将 query 应用到 input table 中新增的数据上,有时候还需要和之前的静态数据一起组合成结果。query 产生的结果成为 Result Table,我们可以选择将 Result Table 输出到外部存储。输出模式有三种:

    • Complete mode:Result Table 全量输出
    • Append mode (default):只有 Result Table 中新增的行才会被输出,所谓新增是指自上一次 trigger 的时候。因为只是输出新增的行,所以如果老数据有改动就不适合使用这种模式。
    • Update mode:只要更新的 Row 都会被输出,相当于 Append mode 的加强版。

    和 batch 模式相比,streaming 模式还提供了一些特有的算子操作,比如 window、watermark、stateful operator 等。

    window,下图是一个基于 event-time 统计 window 内事件的例子。

    import spark.implicits._
    val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }
    // Group the data by window and word and compute the count of each group
    val windowedCounts = words.groupBy(
    window("eventTime", "10 minutes", "5 minutes"),
    $"word"
    ).count()

    如下图所示,窗口大小为 10 分钟,每 5 分钟 trigger 一次。在 12:11 时候收到了一条 12:04 的数据,也就是 late data (什么叫 late data 呢?就是 Processing Time 比 Event Time 要晚),然后去更新其对应的 Result Table 的记录。

    watermark,是也为了处理 ,很多情况下对于这种 late data 的时效数据并没有必要一直保留太久。比如说,数据晚了 10 分钟或者还有点有,但是晚了 1 个小时就没有用了,另外这样设计还有一个好处就是中间状态没有必要维护那么多。watermark 的形式化定义为 max(eventTime) - threshold,早于 watermark 的数据直接丢弃。

    import spark.implicits._
    val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }
    // Group the data by window and word and compute the count of each group
    val windowedCounts = words
    .withWatermark("eventTime", "10 minutes")
    .groupBy(
    window("eventTime", "10 minutes", "5 minutes"),
    $"word")
    .count()

    用下图表示更加形象。在 12:15 trigger 时 watermark 为 12:14 - 10m = 12:04,所以 late date (12:08, dog; 12:13, owl) 都被接收了。在 12:20 trigger 时 watermark 为 12:21 - 10m = 12:11,所以 late data (12:04, donkey) 都丢弃了。

    除此之后 Structured Streaming 还提供了用户可以自定义状态计算逻辑的算子:

    • mapGroupsWithState
    • flatMapGroupsWithState

    看名字大概也能看出来 mapGroupsWithState 是 one -> one,flatMapGroupsWithState 是 one -> multi。这两个算子的底层都是基于 Spark Streaming 的 updateStateByKey。

    4. Continuous Processing Mode

    好,终于要介绍到“真正”的流处理了,我之所以说“真正”是因为 continuous mode 是传统的流处理模式,通过运行一个 long-running 的 operator 用来处理数据。之前 Spark 是基于 micro-batch 模式的,就被很多人诟病不是“真正的”流式处理。continuous mode 这种处理模式只要一有数据可用就会进行处理,如下图所示。epoch 是 input 中数据被发送给 operator 处理的最小单位,在处理过程中,epoch 的 offset 会被记录到 wal 中。另外 continuous 模式下的 snapshot 存储使用的一致性算法是 Chandy-Lamport 算法。

    这种模式相比与 micro-batch 模式缺点和优点都很明显。

    • 缺点是不容易做扩展
    • 优点是延迟更低

    关于为什么延迟更低,下面两幅图可以做到一目了然。

    5. 一致性语义

    对于 Structured Streaming 来说,因为有两种模式,所以我们分开讨论。

    micro-batch 模式 可以提供 end-to-end 的 exactly-once 语义。原因是因为在 input 端和 output  端都做了很多工作来进行保证,比如 input 端 replayable + wal,output 端写入幂等。

    continuous 模式 只能提供 at-least-once 语义。关于 continuous mode 的官方讨论的实在太少,甚至只是提了一下。在和 @李呈祥 讨论之后觉得应该还是 continuous mode 由于要尽可能保证低延迟,所以在 sink 端没有做一致性保证。

    6. Benchmark

        Structured Streaming 的官方论文里面给出了 Yahoo! Streaming Benchmark 的结果,Structured Streaming 的 throughput 大概是 Flink 的 2 倍和 Kafka Streaming 的 90 多倍。

    7. 总结

    总结一下,Structured Streaming 通过提供一套 high-level 的 declarative API 使得流式计算的编写相比 Spark Streaming 简单容易不少,同时通过提供 end-to-end 的 exactly-once 语义。

    8. 闲扯

    最后,闲扯一点别的。Spark 在多年前推出基于 micro-batch 模式的 Spark Streaming 必然是基于当时 Spark Engine 最快的方式,尽管不是真正的流处理,但是在吞吐量更重要的年代,还是尝尽了甜头。而 Spark 的真正基于 continuous 处理模式的 Structured Streaming 直到 Spark 2.3 版本才真正推出,从而导致近两年让 Flink 尝尽了甜头(当然和 Flink 的优秀的语义模型存在很大的关系)。

    在实时计算领域,目前来看,两家的方向都是朝着 Google DataFlow 的方向。由 Spark 的卓越核心 SQL Engine 助力的 Structured Streaming,还是风头正劲的 Flink,亦或是其他流处理引擎,究竟谁将占领统治地位,还是值得期待一下的。

    展开全文
  • Spark Streaming

    千次阅读 2020-04-22 11:11:15
    什么是Spark Streaming 什么是DStream 阐明RDD、DataFrame、DataSet、DStream数据抽象之间的关系。 SparkStreaming代码过程 窗口宽度和滑动距离的关系 0.8版本SparkStreaming集成kafka的差异 Receiver接收方式...
  • File "/opt/cloudera/parcels/CDH-5.7.2-1.cdh5.7.2.p0.18/lib/spark/python/lib/pyspark.zip/pyspark/streaming/context.py", line 289, in awaitTermination File "/opt/cloudera/parcels/CDH-5.7.2-1.cdh5.7.2....
  • 感谢原文链接:http://why-not-learn-something.blogspot.hk/2016/05/apache-spark-streaming-how-to-do.html由于英文水平有限,加之...Apache Spark Streaming : How to do Graceful ShutdownIn my current proje...
  • hive streaming

    万次阅读 2016-05-10 20:40:46
    1. hive 的streaming 包括:  map(), reduce(), transform(), 常用的为transform 2. 恒等变换 select transform(name, salary) using "/bin/cat" as new_name, new_salary from employees where...
  • Streaming Benchmark是Yahoo的一个团队在2015年对当前热门的流式计算平台:Sparking Streaming, Storm和Flink开发的一个基准测试系统。 该系统是当时第一个将这三个流式计算平台在模拟真实应用场景下的基准测试...
  • Streaming Systems高清

    2019-01-28 11:19:51
    彩色,超清晰的pdf,这是一本非常好的流式计算的书,对于当今最流行的流式计算相关的技术理论进行了讲解,谷歌大牛的书,是非常受用的,Flink等计算引擎都是基于此理论
  • Spark Streaming编程指南

    千次阅读 2016-10-13 22:03:35
    本文基于Spark Streaming Programming Guide原文翻译, 加上一些自己的理解和小实验的结果。  一、概述 Spark Streaming是基于Core Spark API的可扩展,高吞吐量,并具有容错能力的用于处理实时数据流的一个组件...
  • sparkStreaming集成Kafka

    千次阅读 2016-12-08 22:48:18
    这几天看了spark集成Kafka,消费Kafka数据并向Kafka发送数据,仿照官方样例写了两个小例子。在此分享一下。 1.添加Kafka的repository 2.DirectKafkaWordCountDemo代码展示 3.kafkaProducer代码展示 ...
  • sparkstreaming

    万次阅读 2017-10-14 12:54:28
    Spark Streaming实时计算框架介绍   随着大数据的发展,人们对大数据的处理要求也越来越高,原有的批处理框架MapReduce适合离线计算,却无法满足实时性要求较高的业务,如实时推荐、用户行为分析等。 Spark ...
  • Structured Streaming 作为 Spark 家族的新成员,通过 Spark SQL/DataFrame 来处理 Batch/Streaming 数据,基本的 SparkSQL API 即可实现离线处理和流式处理,大大的方便了流式计算的开发,另外还提供了丰富的功能。...
  • Spark Streaming基础简介

    2018-07-30 17:42:44
    批量计算和流计算 首先什么是静态数据和流数据?很多企业为了支持决策分析而构建的数据仓库系统,其中存放的大量历史数据就是静态数据。而流数据即数据以大量、快速、时变的流形式持续到达。 ...
  • spark streaming实例编写

    千次阅读 2016-09-02 18:31:46
    运行第一个SparkStreaming程序(及过程中问题解决) Windows下IntelliJ IDEA中调试Spark Standalone sbt-assembly 发布 Scala 项目 使用IDEA开发及测试Spark的环境搭建及简单测试 基于spark运行scala程序...
  • Darwin Streaming Server编译

    千次阅读 2015-07-24 14:52:16
    EasyDarwin是我们在Darwin Streaming Server的基础上进行开发和扩展的,Windows/Linux编译,包括64位编译补丁,我们都进行了整理和测试,后续还会不断进行迭代和扩展,包括整体架构;EasyDarwin网站:...
  • SparkStreaming

    2019-04-14 13:16:06
    SparkStreaming 概述 Spark Streaming可以轻松构建可扩展的容错流应用程序 1)便于使用 2) 容错 3)Spark集成 计算框架 log-》flume-》kafka-》sparkStreaming-》hdfs || mysql -》hive||hbase Dstream 数据...
  • 假设我的通过Spark Streaming 来分析用户实时提交过来的数据,数据包含, time,等其他信息>, 假设.对应 spark代码为 ssc.socketTextStream("...",port) 然后,我想实现基于用户ID的查询 var qid = “u_123“//从...
  • structure streaming

    千次阅读 2020-03-04 21:47:40
    structure steaming在spark streaming上进行了全新架构,持续处理模式支持很快很快,微批处理模式支持毫秒响应,select where groupBy map filter flatMap操作都支持 支持sparkSQL,数据抽象是dataframe 和DataSet ...
  • Structred StreamingStreaming Query分析

    千次阅读 2016-12-28 10:41:52
    Structred StreamingStreaming Query分析 在用户的应用程序中,用户会调用DataStreamWriter.start()方法发起一个Streaming query。 在DataStreamWriter中,会调用df.sparkSession.sessionState....
  • spark streaming

    2018-08-03 10:39:59
    Spark Streaming 课程目标 掌握Spark Streaming的原理 熟练使用Spark Streaming完成流式计算任务 Spark Streaming介绍 Spark Streaming概述 什么是Spark Streaming Spark Streaming类似于Apache ...
  • Hive Streaming

    千次阅读 2016-01-20 17:50:15
    1.Hive Streaming介绍 在前面我们看到了UDF、UDTF、UDAF的实现并不是很简单,而且还要求对Java比较熟悉,而Hive设计的初衷是方便那些非Java人员使用。因此,Hive提供了另一种数据处理方式——Streaming,这样就可以...
  • Hadoop Streaming

    千次阅读 2011-12-01 10:46:55
    Hadoop版本:Hadoop-0.20.204 Hadoop的Streaming框架...Streaming的原理是用Java实现一个包装用户程序的MapReduce程序,该程序负责调用MapReduce Java接口获取key/value对输入,创建一个新的进程启动包装的用户程
  • struct streaming

    2018-10-22 13:21:40
    https://blog.csdn.net/dongyunlon/article/details/79037366 详细说:水印与window函数 https://blog.csdn.net/lovechendongxing/article/details/81748553?utm_source=blogxgwz6

空空如也

1 2 3 4 5 ... 20
收藏数 33,774
精华内容 13,509
关键字:

streaming