Spark之SparkStreaming理论篇

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>sparkstreaming</artifactId>
        <groupId>com.hph</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>helloworld</artifactId>
    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
        </dependency>
    </dependencies>

</project>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}


object StreamingWordCount extends App {
  //创建配置
  val sparkConf = new SparkConf().setAppName("streaming word count").setMaster("local[*]")
  //创建StreamingContext
  val ssc = new StreamingContext(sparkConf, Seconds(5))
  //从socket接口数据
  val lineDStream = ssc.socketTextStream("datanode1", 9999)

  val wordDStream = lineDStream.flatMap(_.split(" "))
  val word2CountDStream = wordDStream.map((_, 1))
  val result = word2CountDStream.reduceByKey(_ + _)


  result.print()

  //启动
  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
31
32
33
34
35
36
37
38
39
40
41
42
43

def start(): Unit = synchronized {
   state match {
     case INITIALIZED =>
       startSite.set(DStream.getCreationSite())
       StreamingContext.ACTIVATION_LOCK.synchronized {
         StreamingContext.assertNoOtherContextIsActive()
         try {
           validate()

           // Start the streaming scheduler in a new thread, so that thread local properties
           // like call sites and job groups can be reset without affecting those of the
           // current thread.
           ThreadUtils.runInNewThread("streaming-start") {
             sparkContext.setCallSite(startSite.get)
             sparkContext.clearJobGroup()
             sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "false")
             savedProperties.set(SerializationUtils.clone(sparkContext.localProperties.get()))
             scheduler.start()
           }
           state = StreamingContextState.ACTIVE
         } catch {
           case NonFatal(e) =>
             logError("Error starting the context, marking it as stopped", e)
             scheduler.stop(false)
             state = StreamingContextState.STOPPED
             throw e
         }
         StreamingContext.setActiveContext(this)
       }
       logDebug("Adding shutdown hook") // force eager creation of logger
       shutdownHookRef = ShutdownHookManager.addShutdownHook(
         StreamingContext.SHUTDOWN_HOOK_PRIORITY)(stopOnShutdown)
       // Registering Streaming Metrics at the start of the StreamingContext
       assert(env.metricsSystem != null)
       env.metricsSystem.registerSource(streamingSource)
       uiTab.foreach(_.attach())
       logInfo("StreamingContext started")
     case ACTIVE =>
       logWarning("StreamingContext has already been started")
     case STOPPED =>
       throw new IllegalStateException("StreamingContext has already been stopped")
   }
 }