- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

Spark基础学习笔记33：Spark Streaming数据源

howard2005 发表于 2022/05/20 01:01:08 2022/05/20

【摘要】文章目录零、本讲学习目标一、基本数据源（一）文件流1、读取文件流概述2、读取文件流演示零、本讲学习目标掌握Spark Streaming基本数据源掌握Spark Strea...

文章目录

零、本讲学习目标
一、基本数据源
- （一）文件流
- - 1、读取文件流概述
  - 2、读取文件流演示

零、本讲学习目标

掌握Spark Streaming基本数据源
掌握Spark Streaming高级数据源

一、基本数据源

StreamingContext API中直接提供了对一些数据源的支持，例如文件系统、Socket连接、RDD队列流等，此类数据源称为基本数据源。

（一）文件流

1、读取文件流概述

对于从任何与HDFS API（HDFS、S3、NFS等）兼容的文件系统上的文件中读取数据，创建DStream的方式：streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)，Spark Streaming将监视目录dataDirectory并处理在该目录中的所有文件。
对于简单的文本文件，创建DStream的方式：streamingContext.textFileStream(dataDirectory)
需要注意的是，文件流不需要运行Receiver，因此不需要为接收文件数据分配CPU内核。

2、读取文件流演示

在HDFS上创建监测目录/stream
待会儿需要将/park目录里的words.txt与test.txt文件拷到监测目录/stream
创建Maven项目 - SparkStreamingDataSourceDemo
将java目录改成scala目录
创建日志属性文件 - log4j.properties

log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spark.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

  
 
  1
  2
  3
  4
  5
  6
  7
  8

添加相关依赖和构建插件

<?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">
    <modelVersion>4.0.0</modelVersion>

    <groupId>net.hw.datasource</groupId>
    <artifactId>SparkStreamingDataSourceDemo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.12</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>
    </dependencies>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.3.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.3.2</version>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</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
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62
  63
  64
  65
  66
  67
  68
  69
  70
  71
  72
  73
  74
  75

创建net.hw.datasource包
在net.hw.datasource包里创建FileWordCount单例对象

package net.hw.datasource

import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 功能：基于文件进行词频统计
 * 作者：华卫
 * 日期：2022年05月18日
 */
object FileWordCount {
  def main(args: Array[String]): Unit = {
    // 创建Spark配置对象
    val conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("FileWordCount")

    // 按照时间间隔为3秒钟切分数据流
    val ssc = new StreamingContext(conf, Seconds(3))
    // 创建行分段流，接收文件流
    val lines = ssc.fileStream[LongWritable, Text, TextInputFormat]("hdfs://master:9000/stream")
    // 生成单词分段流
    val words = lines.flatMap(_._2.toString.split(" "))
    // 计算每一批次中的每个单词数量
    val pairs = words.map((_, 1))
    // 进行词频统计
    val wc = pairs.reduceByKey(_ + _)
    // 输出分段流中每个RDD的词频统计结果
    wc.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