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

举报
howard2005 发表于 2022/05/20 01:01:08 2022/05/20
【摘要】 文章目录 零、本讲学习目标一、基本数据源(一)文件流1、读取文件流概述2、读取文件流演示 零、本讲学习目标 掌握Spark Streaming基本数据源掌握Spark Strea...

零、本讲学习目标

  1. 掌握Spark Streaming基本数据源
  2. 掌握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.txttest.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
  • 启动程序之后,将HDFS/park目录里的words.txttest.txt文件拷贝到/stream目录
    在这里插入图片描述

  • 过一会儿,停止程序,查看控制台输出信息
    在这里插入图片描述

  • 先对words.txt文件进行了词频统计,再对test.txt文件进行了词频统计

  • 注意:第一个文件词频统计结果默认显示前10条,第二个文件词频统计结果只有9条

  • 删除/stream目录里的文件
    在这里插入图片描述

  • 修改FileWordCount,不用通用文件流fileStream,而用文本文件流textFileStream
    在这里插入图片描述

  • 启动程序后,将HDFS/park目录里的test.txtwords.txt文件拷贝到/stream目录
    在这里插入图片描述

  • 停止程序,查看控制台输出信息
    在这里插入图片描述

  • 先对test.txt文件进行了词频统计,再对words.txt文件进行了词频统计

文章来源: howard2005.blog.csdn.net,作者:howard2005,版权归原作者所有,如需转载,请联系作者。

原文链接:howard2005.blog.csdn.net/article/details/124835239

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。