spark streaming

心已赠人 2022-03-25 03:56 213阅读 0赞

<table> 
 <tbody> 
 <tr> 
 <td>问题导读 1.为什么使用spark streaming？ 2.什么是StreamingContext？ 3.什么是DStream？ <img alt="" height="35" src="http://www.aboutyun.com/static/image/hrline/4.gif" width="500"> spark streaming介绍 Spark streaming是Spark核心API的一个扩展，它对实时流式数据的处理具有可扩展性、高吞吐量、可容错性等特点。我们可以从kafka、flume、witter、 ZeroMQ、Kinesis等源获取数据，也可以通过由 高阶函数map、reduce、join、window等组成的复杂算法计算出数据。最后，处理后的数据可以推送到文件系统、数据库、实时仪表盘中 <img alt="" src="http://www.aboutyun.com/data/attachment/forum/201703/03/092932w2572i350ttzti3j.jpg" width="600">&nbsp; 为什么使用spark streaming 很多大数据应用程序需要实时处理数据流。思考： 我们知道spark和storm都能处理实时数据，可是spark是如何处理实时数据的，spark包含比较多组件：包括 
 <ul style="margin-left:14px;"> 
 <li>spark core</li> 
 <li>Spark SQL</li> 
 <li>Spark Streaming</li> 
 <li>GraphX</li> 
 <li>MLlib</li> 
 </ul> spark core中包含RDD、DataFrame和DataSet等，因此spark sql是为了兼容hive而产生的sql语句，GraphX提供的分布式图计算框架，MLlib提供的机器学习框架。因此spark所谓的实时处理数据则是通过spark streaming来实现的。 那么spark有哪些应用spark streaming应用程序可以实时跟踪页面统计，训练机器学习模型或则自动检测异常等 如网站监控 <img alt="" src="http://www.aboutyun.com/data/attachment/forum/201703/03/091320qssfff9z4l3t0l16.png" width="600">&nbsp; 欺诈检测 <img alt="" src="http://www.aboutyun.com/data/attachment/forum/201703/03/091315gxrl9r0krriii5zi.png" width="500">&nbsp; 实时准确数据转移 <img alt="" src="http://www.aboutyun.com/data/attachment/forum/201703/04/165116hyyeji7yijyijj8j.jpg" width="600">&nbsp; 反作弊 + 计费 <img alt="" src="http://www.aboutyun.com/data/attachment/forum/201703/04/165209rhf4qf1lffq3cf44.jpg" width="600">&nbsp; 什么是StreamingContext 为了初始化Spark Streaming程序，一个StreamingContext对象必需被创建，它是Spark Streaming所有流操作的主要入口。一个StreamingContext 对象可以用SparkConf对象创建。StreamingContext这里可能不理解，其实跟SparkContext也差不多的。（可参考<a href="http://www.aboutyun.com/forum.php?mod=viewthread&amp;tid=21018" rel="nofollow">让你真正理解什么是SparkContext, SQLContext 和HiveContext</a>）。同理也有hadoop Context，它们都是全文对象，并且会获取配置文件信息。那么配置文件有哪些？比如hadoop的core-site.xml,hdfs-site.xml等，spark如spark-defaults.conf等。这时候我们可能对StreamingContext有了一定的认识。下面一个例子 为了初始化Spark Streaming程序，一个StreamingContext对象必需被创建，它是Spark Streaming所有流操作的主要入口。 一个StreamingContext 对象可以用SparkConf对象创建。 &nbsp; [Scala]&nbsp;纯文本查看&nbsp;复制代码 <a href="http://www.aboutyun.com/#" rel="nofollow">?</a> 
 <table> 
 <tbody> 
 <tr> 
 <td style="text-align:left;vertical-align:baseline;"> 1 2 3 4 </td> 
 <td style="text-align:left;vertical-align:baseline;"> <code>import</code> <code>org.apache.spark.</code><code>_</code> <code>impoty org.apache.spark.streaming.</code><code>_</code> <code>val</code> <code>conf </code><code>=</code> <code>new</code> <code>SparkConf().setAppName(appName).setMaster(master)</code> <code>val</code> <code>ssc</code><code>=</code><code>new</code> <code>StreamingContext(conf,Seconds(</code><code>1</code><code>))</code> </td> 
 </tr> 
 </tbody> 
 </table> appName表示你的应用程序显示在集群UI上的名字,master 是一个Spark、Mesos、YARN集群URL 或者一个特殊字符串“local”，它表示程序用本地模式运行。当程序运行在集群中时，你并不希望在程序中硬编码 master ，而是希望用 sparksubmit启动应用程序，并从 spark-submit 中得到 master 的值。对于本地测试或者单元测试，你可以传递“local”字符串在同 一个进程内运行Spark Streaming。需要注意的是，它在内部创建了一个SparkContext对象，你可以通过 ssc.sparkContext访问这个SparkContext对象。 批时间片需要根据你的程序的潜在需求以及集群的可用资源来设定，你可以在性能调优那一节获取详细的信息.可以利用已经存在的 SparkContext 对象创建 StreamingContext 对象。 [Scala]&nbsp;纯文本查看&nbsp;复制代码 <a href="http://www.aboutyun.com/#" rel="nofollow">?</a> 
 <table> 
 <tbody> 
 <tr> 
 <td style="text-align:left;vertical-align:baseline;"> 1 2 3 </td> 
 <td style="text-align:left;vertical-align:baseline;"> <code>import</code> <code>org.apache.spark.streaming.</code><code>_</code> <code>val</code> <code>sc </code><code>=</code> <code>... </code><code>// existing SparkContext</code> <code>val</code> <code>ssc </code><code>=</code> <code>new</code> <code>StreamingContext(sc, Seconds(</code><code>1</code><code>))</code> </td> 
 </tr> 
 </tbody> 
 </table> 当一个上下文（context）定义之后，你必须按照以下几步进行操作 &nbsp; 
 <ul style="margin-left:14px;"> 
 <li>定义输入源；</li> 
 <li>准备好流计算指令；</li> 
 <li>利用 streamingContext.start() 方法接收和处理数据；</li> 
 <li>处理过程将一直持续，直到 streamingContext.stop() 方法被调用。</li> 
 </ul> 几点需要注意的地方： 
 <ul style="margin-left:14px;"> 
 <li>一旦一个context已经启动，就不能有新的流算子建立或者是添加到context中。</li> 
 <li>一旦一个context已经停止，它就不能再重新启动</li> 
 <li>在JVM中，同一时间只能有一个StreamingContext处于活跃状态</li> 
 <li>在StreamingContext上调用 stop() 方法，也会关闭SparkContext对象。如果只想仅关闭StreamingContext对象，设</li> 
 <li>置 stop() 的可选参数为false</li> 
 <li>一个SparkContext对象可以重复利用去创建多个StreamingContext对象，前提条件是前面的StreamingContext在后面</li> 
 <li>StreamingContext创建之前关闭（不关闭SparkContext）。</li> 
 </ul> 什么是DStream Spark Streaming支持一个高层的抽象，叫做离散流( discretized stream )或者 DStream ，它代表连续的数据流。DStream既可以利用从Kafka, Flume和Kinesis等源获取的输入数据流创建，也可以 在其他DStream的基础上通过高阶函数获得。在内部，DStream是由一系列RDDs组成。 举例： 一个简单的基于Streaming的workCount代码如下： &nbsp; [Scala]&nbsp;纯文本查看&nbsp;复制代码 <a href="http://www.aboutyun.com/#" rel="nofollow">?</a> 
 <table> 
 <tbody> 
 <tr> 
 <td style="text-align:left;vertical-align:baseline;"> 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 </td> 
 <td style="text-align:left;vertical-align:baseline;"> <code>package</code> <code>com.debugo.example</code> <code>import</code> <code>org.apache.spark.streaming.{Seconds, StreamingContext}</code> <code>import</code> <code>org.apache.spark.streaming.StreamingContext.</code><code>_</code> <code>import</code> <code>org.apache.spark.SparkConf</code> <code>&nbsp;</code>&nbsp; <code>object</code> <code>WordCountStreaming { </code> <code>&nbsp;&nbsp;</code><code>def</code> <code>main(args</code><code>:</code> <code>Array[String])</code><code>:</code> <code>Unit </code><code>=</code><code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code> <code>sparkConf </code><code>=</code> <code>new</code> <code>SparkConf().setAppName(</code><code>"HDFSWordCount"</code><code>).setMaster(</code><code>"spark://172.19.1.232:7077"</code><code>)</code> <code>&nbsp;</code>&nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//create the streaming context</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code>&nbsp; <code>ssc </code><code>=</code> <code>new</code> <code>StreamingContext(sparkConf, Seconds(</code><code>30</code><code>))</code> <code>&nbsp;</code>&nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//process file when new file be found.</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code> <code>lines </code><code>=</code> <code>ssc.textFileStream(</code><code>"file:///home/spark/data"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code> <code>words </code><code>=</code> <code>lines.flatMap(</code><code>_</code><code>.split(</code><code>" "</code><code>))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code> <code>wordCounts </code><code>=</code> <code>words.map(x </code><code>=</code><code>&gt; (x, </code><code>1</code><code>)).reduceByKey(</code><code>_</code> <code>+ </code><code>_</code><code>)</code><code>//这里不是rdd,而是dstream</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>wordCounts.print()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>ssc.start()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>ssc.awaitTermination()</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
 </table> 这段代码实现了当指定的路径有新文件生成时，就会对这些文件执行wordcount，并把结果print。具体流程如下： <img alt="" height="164" src="http://www.aboutyun.com/data/attachment/forum/201603/06/141658v0gkdxar0degb22g.png" width="592"> 代码诠释： 使用Spark Streaming就需要创建StreamingContext对象（类似SparkContext）。创建StreamingContext对象所需的参数与SparkContext基本一致，包括设定Master节点(setMaster），设定应用名称(setAppName)。第二个参数Seconds(30)，指定了Spark Streaming处理数据的时间间隔为30秒。需要根据具体应用需要和集群处理能力进行设置。 val lines = ssc.textFileStream("file:///home/spark/data")为创建lines Dstream val words = lines.flatMap(_.split(" "))为通过flatMap转换为words Dstream 我们在引一例，比如创建Twitter val tweets=ssc.twitterStream() <img alt="" src="http://www.aboutyun.com/data/attachment/forum/201703/03/094510b996zw0tncikgskn.jpg" width="600">&nbsp; 其中为tweets为DStream 后续更新 转载注明来自：about云（<a href="http://www.aboutyun.com/" rel="nofollow">www.aboutyun.com</a>）</td> 
 </tr> 
 </tbody> 
</table>