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

Posted howard2005

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark基础学习笔记33:Spark Streaming数据源相关的知识,希望对你有一定的参考价值。

文章目录

零、本讲学习目标

  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
  • 添加相关依赖和构建插件
<?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>
  • 创建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()
  

  • 启动程序之后,将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文件进行了词频统计

以上是关于Spark基础学习笔记33:Spark Streaming数据源的主要内容,如果未能解决你的问题,请参考以下文章

2022年Spark基础学习笔记目录

Spark基础学习笔记01:初步了解Spark

Spark基础学习笔记32:Spark Streaming概述

Spark基础学习笔记06:搭建Spark On YARN模式的集群

Spark基础学习笔记03:Spark运行架构及原理

Spark基础学习笔记02:搭建Spark环境