Spark Dstream 创建

时间：2019-07-09 19:30:00 阅读：98 评论：0 收藏：0 [点我收藏+]

标签：name data 格式映射 param stream table NPU done

第 3 章 Dstream 创建

　　Spark Streaming 原生支持一些不同的数据源。一些“核心”数据源已经被打包到 Spark

Streaming 的 Maven 工件中，而其他的一些则可以通过 spark-streaming-kafka 等附加工件获取。

每个接收器都以 Spark 执行器程序中一个长期运行的任务的形式运行，因此会占据分配给应用

的 CPU 核心。此外，我们还需要有可用的 CPU 核心来处理数据。这意味着如果要运行多个接

收器，就必须至少有和接收器数目相同的核心数，还要加上用来完成计算所需要的核心数。例如，

如果我们想要在流计算应用中运行 10 个接收器，那么至少需要为应用分配 11 个 CPU 核心。

所以如果在本地模式运行，不要使用 local 或者 local[1]。

3.1 文件数据源

3.1.1 用法及说明

　　文件数据流：能够读取所有 HDFS API 兼容的文件系统文件，通过 fileStream 方法进行读取，

Spark Streaming 将会监控 dataDirectory 目录并不断处理移动进来的文件，记住目前不支持嵌套目

录。

streamingContext.textFileStream(dataDirectory)

注意事项：

　　1）文件需要有相同的数据格式；

　　2）文件进入 dataDirectory 的方式需要通过移动或者重命名来实现；

　　3）一旦文件移动进目录，则不能再修改，即便修改了也不会读取新数据；

3.1.2 案例实操

（1）在 HDFS 上建好目录

[lxl@hadoop102 spark]$ hadoop fs -mkdir /fileStream

（2）在/opt/module/data 创建三个文件

[lxl@hadoop102 data]$ touch a.tsv
[lxl@hadoop102 data]$ touch b.tsv
[lxl@hadoop102 data]$ touch c.tsv

添加如下数据：
Helloatguigu
Hellospark

（3）编写代码

package com.lxl
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream
object FileStream {
  def main(args: Array[String]): Unit = {
    //1.初始化 Spark 配置信息
    val sparkConf = new SparkConf().setMaster("local[*]")
      .setAppName("StreamWordCount")
    //2.初始化 SparkStreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //3.监控文件夹创建 DStream
    val dirStream = ssc.textFileStream("hdfs://hadoop102:9000/fileStream")
    //4.将每一行数据做切分，形成一个个单词
    val wordStreams = dirStream.flatMap(_.split("\t"))
    //5.将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))
    //6.将相同的单词次数做统计
    val wordAndCountStreams = wordAndOneStreams.reduceByKey(_ + _)
    //7.打印
    wordAndCountStreams.print()
    //8.启动 SparkStreamingContext
    ssc.start()
    ssc.awaitTermination()
  }
}

（4）启动程序并向 fileStream 目录上传文件

[lxl@hadoop102 data]$ hadoop fs -put ./a.tsv /fileStream
[lxl@hadoop102 data]$ hadoop fs -put ./b.tsv /fileStream
[lxl@hadoop102 data]$ hadoop fs -put ./c.tsv /fileStream

（5）获取计算结果

-------------------------------------------
Time: 1539073810000 ms
-------------------------------------------
-------------------------------------------
Time: 1539073815000 ms
-------------------------------------------
(Hello,4)
(spark,2)
(atguigu,2)
-------------------------------------------
Time: 1539073820000 ms
-------------------------------------------
(Hello,2)
(spark,1)
(atguigu,1)
-------------------------------------------
Time: 1539073825000 ms
-------------------------------------------

3.2 RDD 队列

3.2.1 用法及说明

测试过程中，可以通过使用 ssc.queueStream(queueOfRDDs)来创建 DStream，每一个推送到

这个队列中的 RDD，都会作为一个 DStream 处理。

3.2.2 案例实操

1）需求：循环创建几个 RDD，将 RDD 放入队列。通过 SparkStream 创建 Dstream，计算 WordCount

2）编写代码

package com.atguigu
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scala.collection.mutable
object RDDStream {
  def main(args: Array[String]) {
    //1.初始化 Spark 配置信息
    val conf = new SparkConf().setMaster("local[*]").setAppName("RDDStream")
    //2.初始化 SparkStreamingContext
    val ssc = new StreamingContext(conf, Seconds(4))
    //3.创建 RDD 队列
    val rddQueue = new mutable.Queue[RDD[Int]]()
    //4.创建 QueueInputDStream
    val inputStream = ssc.queueStream(rddQueue,oneAtATime = false)
    //5.处理队列中的 RDD 数据
    val mappedStream = inputStream.map((_,1))
    val reducedStream = mappedStream.reduceByKey(_ + _)
    //6.打印结果
    reducedStream.print()
    //7.启动任务
    ssc.start()
    //8.循环创建并向 RDD 队列中放入 RDD
    for (i <- 1 to 5) {
      rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }
    ssc.awaitTermination()
  }
}

3）结果展示

-------------------------------------------
Time: 1539075280000 ms
-------------------------------------------
(4,60)
(0,60)
(6,60)
(8,60)
(2,60)
(1,60)
(3,60)
(7,60)
(9,60)
(5,60)
-------------------------------------------
Time: 1539075284000 ms
-------------------------------------------
(4,60)
(0,60)
(6,60)
(8,60)
(2,60)
(1,60)
(3,60)
(7,60)
(9,60)
(5,60)
-------------------------------------------
Time: 1539075288000 ms
-------------------------------------------
(4,30)
(0,30)
(6,30)
(8,30)
(2,30)
(1,30)
(3,30)
(7,30)
(9,30)
(5,30)
-------------------------------------------
Time: 1539075292000 ms
-------------------------------------------

3.3 自定义数据源

3.3.1 用法及说明

需要继承 Receiver，并实现 onStart、onStop 方法来自定义数据源采集。

3.3.2 案例实操

1）需求：自定义数据源，实现监控某个端口号，获取该端口号内容。

2）自定义数据源

package com.lxl

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket
import java.nio.charset.StandardCharsets
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

class CustomerReceiver(host: String, port: Int) extends
  Receiver[String](StorageLevel.MEMORY_ONLY) {
  //最初启动的时候，调用该方法，作用为：读数据并将数据发送给 Spark
  override def onStart(): Unit = {
    new Thread("Socket Receiver") {
      override def run() {
        receive()
      }
    }.start()
  }
  //读数据并将数据发送给 Spark
  def receive(): Unit = {
    //创建一个 Socket
    var socket: Socket = new Socket(host, port)
    //定义一个变量，用来接收端口传过来的数据
    var input: String = null
    //创建一个 BufferedReader 用于读取端口传来的数据
    val reader = new BufferedReader(new InputStreamReader(socket.getInputStream,
      StandardCharsets.UTF_8))
    //读取数据
    input = reader.readLine()
    //当 receiver 没有关闭并且输入数据不为空，则循环发送数据给 Spark
    while (!isStopped() && input != null) {
      store(input)
      input = reader.readLine()
    }
    //跳出循环则关闭资源
    reader.close()
    socket.close()
    //重启任务
    restart("restart")
  }
  override def onStop(): Unit = {}
}

3）使用自定义的数据源采集数据

package com.atguigu

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

object FileStream {
  def main(args: Array[String]): Unit = {
    //1.初始化 Spark 配置信息
    Val sparkConf = new SparkConf().setMaster("local[*]")
      .setAppName("StreamWordCount")
    //2.初始化 SparkStreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //3.创建自定义 receiver 的 Streaming
    val lineStream = ssc.receiverStream(new CustomerReceiver("hadoop102", 9999))
    //4.将每一行数据做切分，形成一个个单词
    val wordStreams = lineStream.flatMap(_.split("\t"))
    //5.将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))
    //6.将相同的单词次数做统计
    val wordAndCount = wordAndOneStreams.reduceByKey(_ + _)
    //7.打印
    wordAndCountStreams.print()
    //8.启动 SparkStreamingContext
    ssc.start()
    ssc.awaitTermination()
  }
}

3.4 Kafka 数据源

3.4.1 用法及说明

在工程中需要引入 Maven 工件 spark- streaming-kafka_2.10 来使用它。包内提供的

KafkaUtils 对象可以在 StreamingContext 和 JavaStreamingContext 中以你的 Kafka 消息创建出

DStream。由于 KafkaUtils 可以订阅多个主题，因此它创建出的 DStream 由成对的主题和消息

组成。要创建出一个流数据，需要使用 StreamingContext 实例、一个由逗号隔开的 ZooKeeper

主机列表字符串、消费者组的名字(唯一名字)，以及一个从主题到针对这个主题的接收器线程数

的映射表来调用 createStream() 方法。

3.4.2 案例实操

1）需求 1：通过 SparkStreaming 从 Kafka 读取数据，并将读取过来的数据做简单计

算(WordCount)，最终打印到控制台。

（1）导入依赖

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka -->
<dependency>
　　<groupId>org.apache.spark</groupId>
　　<artifactId>spark-streaming-kafka_2.11</artifactId>
　　<version>1.6.3</version>
</dependency>

（2）编写代码

package com.lxl
import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
object KafkaSparkStreaming {
  def main(args: Array[String]): Unit = {
    //1.创建 SparkConf 并初始化 SSC
    val sparkConf: SparkConf = new
        SparkConf().setMaster("local[*]").setAppName("KafkaSparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //2.定义 kafka 参数
    val zookeeper = "hadoop102:2181,hadoop103:2181,hadoop104:2181"
    val topic = "source"
    val consumerGroup = "spark"
    //3.将 kafka 参数映射为 map
    val kafkaParam: Map[String, String] = Map[String, String](
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      ConsumerConfig.GROUP_ID_CONFIG -> consumerGroup,
      "zookeeper.connect" -> zookeeper
    )
    //4.通过 KafkaUtil 创建 kafkaDSteam
    val kafkaDSteam: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](
      ssc,
      kafkaParam,
      Map[String, Int](topic -> 3),
      StorageLevel.MEMORY_ONLY
    )
    //5.对 kafkaDSteam 做计算（WordCount）
    kafkaDSteam.foreachRDD {
      rdd => {
        val word: RDD[String] = rdd.flatMap(_._2.split(" "))
        val wordAndOne: RDD[(String, Int)] = word.map((_, 1))
        val wordAndCount: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
        wordAndCount.collect().foreach(println)
      }
    }
    //6.启动 SparkStreaming
    ssc.start()
    ssc.awaitTermination()
  }
}

Spark Dstream 创建

标签：name data 格式映射 param stream table NPU done

原文地址：https://www.cnblogs.com/LXL616/p/11159239.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行