Spark的Streaming和Spark的SQL简单入门学习

1、Spark Streaming是什么?

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

2、Spark与Storm的对比

  a、Spark开发语言:Scala、Storm的开发语言:Clojure。

  b、Spark编程模型:DStream、Storm编程模型:Spout/Bolt。

  c、Spark和Storm的对比介绍:

Spark:

Storm:

 3、什么是DStream?

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

对数据的操作也是按照RDD为单位来进行的:

计算过程由Spark engine来完成

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

  a、Transformations on DStream:

Transformation

Meaning

map(func)

Return a new DStream by passing each element of the source DStream through a function func.

flatMap(func)

Similar to map, but each input item can be mapped to 0 or more output items.

filter(func)

Return a new DStream by selecting only the records of the source DStream on which func returns true.

repartition(numPartitions)

Changes the level of parallelism in this DStream by creating more or fewer partitions.

union(otherStream)

Return a new DStream that contains the union of the elements in the source DStream and otherDStream.

count()

Return a new DStream of single-element RDDs by counting the number of elements in each RDD of the source DStream.

reduce(func)

Return a new DStream of single-element RDDs by aggregating the elements in each RDD of the source DStream using a function func (which takes two arguments and returns one). The function should be associative so that it can be computed in parallel.

countByValue()

When called on a DStream of elements of type K, return a new DStream of (K, Long) pairs where the value of each key is its frequency in each RDD of the source DStream.

reduceByKey(func, [numTasks])

When called on a DStream of (K, V) pairs, return a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. Note: By default, this uses Spark's default number of parallel tasks (2 for local mode, and in cluster mode the number is determined by the config property spark.default.parallelism) to do the grouping. You can pass an optional numTasks argument to set a different number of tasks.

join(otherStream, [numTasks])

When called on two DStreams of (K, V) and (K, W) pairs, return a new DStream of (K, (V, W)) pairs with all pairs of elements for each key.

cogroup(otherStream, [numTasks])

When called on a DStream of (K, V) and (K, W) pairs, return a new DStream of (K, Seq[V], Seq[W]) tuples.

transform(func)

Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream. This can be used to do arbitrary RDD operations on the DStream.

updateStateByKey(func)

Return a new "state" DStream where the state for each key is updated by applying the given function on the previous state of the key and the new values for the key. This can be used to maintain arbitrary state data for each key.

特殊的Transformations

1.UpdateStateByKey Operation
UpdateStateByKey原语用于记录历史记录,上文中Word Count示例中就用到了该特性。若不用UpdateStateByKey来更新状态,那么每次数据进来后分析完成后,结果输出后将不在保存

2.Transform Operation
Transform原语允许DStream上执行任意的RDD-to-RDD函数。通过该函数可以方便的扩展Spark API。此外,MLlib(机器学习)以及Graphx也是通过本函数来进行结合的。

3.Window Operations
Window Operations有点类似于Storm中的State,可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming的允许状态

  b、Output Operations on DStreams:

    Output Operations可以将DStream的数据输出到外部的数据库或文件系统,当某个Output Operations原语被调用时(与RDD的Action相同),streaming程序才会开始真正的计算过程。

Output Operation

Meaning

print()

Prints the first ten elements of every batch of data in a DStream on the driver node running the streaming application. This is useful for development and debugging.

saveAsTextFiles(prefix, [suffix])

Save this DStream's contents as text files. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]".

saveAsObjectFiles(prefix, [suffix])

Save this DStream's contents as SequenceFiles of serialized Java objects. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]".

saveAsHadoopFiles(prefix, [suffix])

Save this DStream's contents as Hadoop files. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]".

foreachRDD(func)

The most generic output operator that applies a function, func, to each RDD generated from the stream. This function should push the data in each RDD to an external system, such as saving the RDD to files, or writing it over the network to a database. Note that the function func is executed in the driver process running the streaming application, and will usually have RDD actions in it that will force the computation of the streaming RDDs.

4、 Spark Streaming的练习使用:

从Socket实时读取数据,进行实时处理,首先测试是否安装nc:

然后检查是否安装:[root@slaver1 hadoop]# which nc

  然后安装nc:[root@slaver1 hadoop]# yum install -y nc(此种方法安装出现错误,不建议使用)

  [root@slaver1 hadoop]# wget   http://vault.centos.org/6.6/os/x86_64/Packages/nc-1.84-22.el6.x86_64.rpm   [root@slaver1 hadoop]# rpm   -iUv    nc-1.84-22.el6.x86_64.rpm

  然后在窗口执行如下命令:[root@slaver1 hadoop]#  nc -lk 9999(输入消息)。

  然后复制这个窗口,执行如下命令:[root@slaver1 hadoop]# nc slaver1 9999(可以接受输入的消息)。

5、开始测试:

[hadoop@slaver1 ~]$ nc -lk 9999

[hadoop@slaver1 spark-1.5.1-bin-hadoop2.4]$ ./bin/run-example streaming.NetworkWordCount 192.168.19.131 9999

然后在第一行的窗口输入例如:hello world hello world hadoop world spark world flume world hello world

看第二行的窗口是否进行计数计算;


1、Spark SQL and DataFrame

a、什么是Spark SQL?
  Spark SQL是Spark用来处理结构化数据的一个模块,它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用。
b、为什么要学习Spark SQL?
  我们已经学习了Hive,它是将Hive SQL转换成MapReduce然后提交到集群上执行,大大简化了编写MapReduce的程序的复杂性,由于MapReduce这种计算模型执行效率比较慢。所有Spark SQL的应运而生,它是将Spark SQL转换成RDD,然后提交到集群执行,执行效率非常快!
c、Spark的特点:
  易整合、统一的数据访问方式、兼容Hive、标准的数据连接。
d、什么是DataFrames?
  与RDD类似,DataFrame也是一个分布式数据容器。然而DataFrame更像传统数据库的二维表格,除了数据以外,还记录数据的结构信息,即schema。同时,与Hive类似,DataFrame也支持嵌套数据类型(struct、array和map)。从API易用性的角度上 看,DataFrame API提供的是一套高层的关系操作,比函数式的RDD API要更加友好,门槛更低。由于与R和Pandas的DataFrame类似,Spark DataFrame很好地继承了传统单机数据分析的开发体验。

2、创建DataFrames?

在Spark SQL中SQLContext是创建DataFrames和执行SQL的入口,在spark-1.5.2中已经内置了一个sqlContext:

1.在本地创建一个文件,有三列,分别是id、name、age,用空格分隔,然后上传到hdfs上
hdfs dfs -put person.txt /

2.在spark shell执行下面命令,读取数据,将每一行的数据使用列分隔符分割
val lineRDD = sc.textFile("hdfs://node1.itcast.cn:9000/person.txt").map(_.split(" "))

3.定义case class(相当于表的schema)
case class Person(id:Int, name:String, age:Int)

4.将RDD和case class关联
val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))

5.将RDD转换成DataFrame
val personDF = personRDD.toDF

6.对DataFrame进行处理
personDF.show

3、DataFrame常用操作:

DSL风格语法
//查看DataFrame中的内容
personDF.show

//查看DataFrame部分列中的内容
personDF.select(personDF.col("name")).show
personDF.select(col("name"), col("age")).show
personDF.select("name").show

//打印DataFrame的Schema信息
personDF.printSchema

//查询所有的name和age,并将age+1
personDF.select(col("id"), col("name"), col("age") + 1).show
personDF.select(personDF("id"), personDF("name"), personDF("age") + 1).show

//过滤age大于等于18的
personDF.filter(col("age") >= 18).show

//按年龄进行分组并统计相同年龄的人数
personDF.groupBy("age").count().show()

 4、SQL风格语法:

如果想使用SQL风格的语法,需要将DataFrame注册成表
personDF.registerTempTable("t_person")

//查询年龄最大的前两名
sqlContext.sql("select * from t_person order by age desc limit 2").show


//显示表的Schema信息
sqlContext.sql("desc t_person").show

待续......

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Spark学习技巧

Spark SQL用UDF实现按列特征重分区

解决问题之前,要先了解一下Spark 原理,要想进行相同数据归类到相同分区,肯定要有产生shuffle步骤。

18610
来自专栏Albert陈凯

3.3RDD的转换和DAG的生成

3.3 RDD的转换和DAG的生成 Spark会根据用户提交的计算逻辑中的RDD的转换和动作来生成RDD之间的依赖关系,同时这个计算链也就生成了逻辑上的DAG...

30970
来自专栏个人分享

最最简单的~WordCount¬

步骤1:textFile先生成HadoopRDD,然后再通过map操作生成MappedRDD.

11510
来自专栏张浩的专栏

Spark Streaming基于网络的词频统计

18110
来自专栏浪淘沙

SparkStreaming编程实现

3.MyNetworkTotalWordCountV2.scala(开发自己的实时词频统计程序(累计单词出现次数))

23150
来自专栏大数据

Zzreal的大数据笔记-SparkDay05

Spark Streaming SparkStreaming部分没做知识点的笔记,直接从代码上理解它的用法。后面整理Storm的时候会与SparkStreami...

23160
来自专栏一名叫大蕉的程序员

Spark你一定学得会(一)No.7

我是小蕉。 上一篇大家说没有干货,妈蛋回南天哪来的干货你告诉我!!!还好这几天天气还不错,干货来了。 首先祭上今天关键代码,要做的事情就是从Hive表中取得年龄...

20750
来自专栏岑玉海

Spark源码系列(五)分布式缓存

这一章想讲一下Spark的缓存是如何实现的。这个persist方法是在RDD里面的,所以我们直接打开RDD这个类。 def persist(newLevel...

40250
来自专栏数据处理

提交任务到集群

16020
来自专栏Spark学习技巧

RDD的join和Dstream的join有什么区别?

Dstream这个类实际上支持的只是Spark Streaming的基础操作算子,比如: map, filter 和window.PairDStreamFunc...

22910

扫码关注云+社区

领取腾讯云代金券