概述

Spark Streaming是Spark Core Api的一个扩展，用于流式数据的处理。支持的数据源有很多，例如：Kafka、Flume、Twitter、ZeroMQ、TCP、HDFS/S3等等。数据输入后可以使用Spark的高度抽象原语进行运算。运算完毕后，将结果存放在如HDFS、关系型数据库等等Sink中。

严格意义上来说，SparkStreaming采用的并不是实时的流处理方式(来一条数据处理一条)。如上图所示，spark Streaming将接收的数据流周期性的将数据分为一个个的数据集(可以理解为RDD)，发送给Spark Engine处理。

idea开发Spark Streaming应用

idea新建maven项目，配置scala环境

File–>Project Structure -->Modules，添加scala依赖库

配置pom.xml

<dependency>
    <groupId>org.apache.sparkgroupId>
    <artifactId>spark-core_2.12artifactId>
    <version>3.3.0version>
dependency>
<dependency>
    <groupId>org.apache.sparkgroupId>
    <artifactId>spark-streaming_2.12artifactId>
    <version>3.3.0version>
dependency>
1
2
3
4
5
6
7
8
9
10

scala源码示例

package com.hjt.yxh.hw.dstream
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object WordCountStreamApp {

  def main(args: Array[String]): Unit = {
    val conf:SparkConf = new SparkConf()
    conf.setMaster("local[*]").setAppName("DStreamTestApp")
    
    //创建一个StreamingContext
    val ssc:StreamingContext = new StreamingContext(conf,Seconds(3))
    
    //创建一个socket文本流
    val line: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.0.52",8888)
    
    //对接收到的数据进行处理(类似于RDD的转换算子操作，也是懒执行的)
    val sensorDs = line
      .filter(_.nonEmpty)
        .flatMap(data=>{data.split(" ")})
        .map((_,1))
    
    sensorDs.reduceByKey(_+_).print()
    
    //启动streamingContext
    ssc.start()
    
    ssc.awaitTermination()
  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

使用nc工具模拟发送数据

[root@k8s-node3 ~]# nc -l -p 8888
hello world
hello world
1
2
3

滚动输出：

22/08/02 17:50:51 INFO DAGScheduler: Job 6 finished: print at WordCountStreamApp.scala:18, took 0.022961 s

-------------------------------------------
Time: 1659433848000 ms
-------------------------------------------
(world,2)
(hello,2)
-------------------------------------------
Time: 1659433851000 ms
-------------------------------------------
1
2
3
4
5
6
7
8
9
10

可能遇到的报错

• Exception in thread “main” java.lang.NoClassDefFoundError: org/apache/spark/SparkConf

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/spark/SparkConf
	at com.hjt.yxh.hw.dstream.WordCountStreamApp$.main(WordCountStreamApp.scala:9)
	at com.hjt.yxh.hw.dstream.WordCountStreamApp.main(WordCountStreamApp.scala)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:64)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:564)
	at com.intellij.rt.execution.application.AppMainV2.main(AppMainV2.java:131)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.SparkConf
	at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:606)
	at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:168)
	at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
1
2
3
4
5
6
7
8
9
10
11
12

因为在官网上拷贝下来的Maven依赖中有provided，记得删掉或者改成compile，然后重新导入，使得更改生效

<dependency>
    <groupId>org.apache.sparkgroupId>
    <artifactId>spark-streaming_2.12artifactId>
    <version>3.3.0version>
    
    <scope>providedscope>
dependency>
1
2
3
4
5
6
7

总结

开发SparkStream应用时有几点需要注意：

• 在本地开发模式下，不要使用conf.setMaster(“local”)或者conf.setMaster(“local[1]”)这种方式设置Master的url，因为这意味着只有一个线程被启动用来运行任务。如果我们是基于一个receiver的方式来使用一个输入流，意味着这个线程将被用来接收数据，而没有多余的线程来处理接收到的数据。

• 同样的逻辑，在基于集群运行sparkStream应用时，分配给Spark Streaming 应用的cores的个数，必须大于分配给receiver的cores的个数，否者系统只能接收数据而不能对这些数据进程处理。