Spark编程实验五：Spark Structured Streaming编程

2024-02-12 22:20
文章标签 编程 实验 spark streaming structured

本文主要是介绍Spark编程实验五：Spark Structured Streaming编程，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

目录

一、目的与要求

二、实验内容

三、实验步骤

1、Syslog介绍

2、通过Socket传送Syslog到Spark

3、Syslog日志拆分为DateFrame

4、对Syslog进行查询

四、结果分析与实验体会

一、目的与要求

1、通过实验掌握Structured Streaming的基本编程方法；
2、掌握日志分析的常规操作，包括拆分日志方法和分析场景。

二、实验内容

1、通过Socket传送Syslog到Spark

        日志分析是一个大数据分析中较为常见的场景。在Unix类操作系统里，Syslog广泛被应用于系统或者应用的日志记录中。Syslog通常被记录在本地文件内，比如Ubuntu内为/var/log/syslog文件名，也可以被发送给远程Syslog服务器。Syslog日志内一般包括产生日志的时间、主机名、程序模块、进程名、进程ID、严重性和日志内容。

        日志一般会通过Kafka等有容错保障的源发送，本实验为了简化，直接将Syslog通过Socket源发送。新建一个终端，执行如下命令：

$ tail -n+1 -f /var/log/syslog | nc -lk 9988

        “tail -n+1 -f /var/log/syslog”表示从第一行开始打印文件syslog的内容。“-f”表示如果文件有增加则持续输出最新的内容。然后，通过管道把文件内容发送到nc程序（nc程序可以进一步把数据发送给Spark）。

        如果/var/log/syslog内的内容增长速度较慢，可以再新开一个终端（计作“手动发送日志终端”），手动在终端输入如下内容来增加日志信息到/var/log/syslog内：

$ logger ‘I am a test error log message.’

2、对Syslog进行查询

由Spark接收nc程序发送过来的日志信息，然后完成以下任务：

（1）统计CRON这个进程每小时生成的日志数，并以时间顺序排列，水印设置为1分钟。
（2）统计每小时的每个进程或者服务分别产生的日志总数，水印设置为1分钟。
（3）输出所有日志内容带error的日志。

三、实验步骤

1、Syslog介绍

        分析日志是一个大数据分析中较为常见的场景。在Unix类操作系统里，Syslog广泛被应用于系统或者应用的日志记录中。Syslog通常被记录在本地文件内，比如Ubuntu内为/var/log/syslog文件名，也可以被发送给远程Syslog服务器。Syslog日志内一般包括产生日志的时间、主机名、程序模块、进程名、进程ID、严重性和日志内容。

2、通过Socket传送Syslog到Spark

        日志一般会通过kafka等有容错保障的源发送，本实验为了简化，直接将syslog通过Socket源发送。新开一个终端，命令为“tail终端”，输入

tail -n+1 -f /var/log/syslog | nc -lk 9988

        tail命令加-n+1代表从第一行开始打印文件内容。-f代表如果文件有增加则持续输出最新的内容。通过管道发送到nc命令起的在本地9988上的服务上。
        如果/var/log/syslog内的内容增长速度较慢，可以再新开一个终端，命名为“手动发送log终端”，手动在终端输入

logger ‘I am a test error log message.’

来增加日志信息到/var/log/syslog内。

3、Syslog日志拆分为DateFrame

        Syslog每行的数据类似以下：

Nov 24 13:17:01 spark CRON[18455]: (root) CMD (cd / && run-parts --report /etc/cron.hourly)

        最前面为时间，接着是主机名，进程名，可选的进程ID，冒号后是日志内容。在Spark内，可以使用正则表达式对syslog进行拆分成结构化字段，以下是示例代码：

 # 定义一个偏应用函数，从固定的pattern获取日志内匹配的字段fields = partial(regexp_extract, str="value", pattern="^(\w{3}\s*\d{1,2} \d{2}:\d{2}:\d{2}) (.*?) (.*?)\[*\d*\]*: (.*)$")words = lines.select(to_timestamp(format_string('2019 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),fields(idx=2).alias("hostname"),fields(idx=3).alias("tag"),fields(idx=4).alias("content"),)

        to_timestamp(format_string('2018 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),这句是对Syslog格式的一个修正，因为系统默认的Syslog日期是没有年的字段，所以使用format_string函数强制把拆分出来的第一个字段前面加上2019年，再根据to_timestamp格式转换成timestamp字段。在接下来的查询应当以这个timestamp作为事件时间。

4、对Syslog进行查询

由Spark接收nc程序发送过来的日志信息，然后完成以下任务。

（1）统计CRON这个进程每小时生成的日志数，并以时间顺序排列，水印设置为1分钟。

        在新开的终端内输入 vi spark_exercise_testsyslog1.py ，贴入如下代码并运行。运行之前需要关闭“tail终端”内的tail命令并重新运行tail命令，否则多次运行测试可能导致没有新数据生成。

#!/usr/bin/env python3from functools import partialfrom pyspark.sql import SparkSession
from pyspark.sql.functions import *if __name__ == "__main__":spark = SparkSession \.builder \.appName("StructuredSyslog") \.getOrCreate()lines = spark \.readStream \.format("socket") \.option("host", "localhost") \.option("port", 9988) \.load()# Nov 24 13:17:01 spark CRON[18455]: (root) CMD (   cd / && run-parts --report /etc/cron.hourly)# 定义一个偏应用函数，从固定的pattern获取日志内匹配的字段fields = partial(regexp_extract, str="value", pattern="^(\w{3}\s*\d{1,2} \d{2}:\d{2}:\d{2}) (.*?) (.*?)\[*\d*\]*: (.*)$")words = lines.select(to_timestamp(format_string('2019 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),fields(idx=2).alias("hostname"),fields(idx=3).alias("tag"),fields(idx=4).alias("content"),)# (1).  统计CRON这个进程每小时生成的日志数，并以时间顺序排列，水印设置为1分钟。windowedCounts1 = words \.filter("tag = 'CRON'") \.withWatermark("timestamp", "1 minutes") \.groupBy(window('timestamp', "1 hour")) \.count() \.sort(asc('window'))# 开始运行查询并在控制台输出query = windowedCounts1 \.writeStream \.outputMode("complete") \.format("console") \.option('truncate', 'false')\.trigger(processingTime="3 seconds") \.start()query.awaitTermination()

（2）统计每小时的每个进程或者服务分别产生的日志总数，水印设置为1分钟。

        在新开的终端内输入 vi spark_exercise_testsyslog2.py ，贴入如下代码并运行。运行之前需要关闭“tail终端”内的tail命令并重新运行tail命令，否则多次运行测试可能导致没有新数据生成。

#!/usr/bin/env python3from functools import partialfrom pyspark.sql import SparkSession
from pyspark.sql.functions import *if __name__ == "__main__":spark = SparkSession \.builder \.appName("StructuredSyslog") \.getOrCreate()lines = spark \.readStream \.format("socket") \.option("host", "localhost") \.option("port", 9988) \.load()# Nov 24 13:17:01 spark CRON[18455]: (root) CMD (   cd / && run-parts --report /etc/cron.hourly)# 定义一个偏应用函数，从固定的pattern获取日志内匹配的字段fields = partial(regexp_extract, str="value", pattern="^(\w{3}\s*\d{1,2} \d{2}:\d{2}:\d{2}) (.*?) (.*?)\[*\d*\]*: (.*)$")words = lines.select(to_timestamp(format_string('2019 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),fields(idx=2).alias("hostname"),fields(idx=3).alias("tag"),fields(idx=4).alias("content"),)# (2).  统计每小时的每个进程或者服务分别产生的日志总数，水印设置为1分钟。windowedCounts2 = words \.withWatermark("timestamp", "1 minutes") \.groupBy('tag', window('timestamp', "1 hour")) \.count() \.sort(asc('window'))# 开始运行查询并在控制台输出query = windowedCounts2 \.writeStream \.outputMode("complete") \.format("console") \.option('truncate', 'false')\.trigger(processingTime="3 seconds") \.start()query.awaitTermination()

（3）输出所有日志内容带error的日志。

        在新开的终端内输入 vi spark_exercise_testsyslog3.py ，贴入如下代码并运行。运行之前需要关闭“tail终端”内的tail命令并重新运行tail命令，否则多次运行测试可能导致没有新数据生成。

#!/usr/bin/env python3from functools import partialfrom pyspark.sql import SparkSession
from pyspark.sql.functions import *if __name__ == "__main__":spark = SparkSession \.builder \.appName("StructuredSyslog") \.getOrCreate()lines = spark \.readStream \.format("socket") \.option("host", "localhost") \.option("port", 9988) \.load()# Nov 24 13:17:01 spark CRON[18455]: (root) CMD (   cd / && run-parts --report /etc/cron.hourly)# 定义一个偏应用函数，从固定的pattern获取日志内匹配的字段fields = partial(regexp_extract, str="value", pattern="^(\w{3}\s*\d{1,2} \d{2}:\d{2}:\d{2}) (.*?) (.*?)\[*\d*\]*: (.*)$")words = lines.select(to_timestamp(format_string('2019 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),fields(idx=2).alias("hostname"),fields(idx=3).alias("tag"),fields(idx=4).alias("content"),)# (3).  输出所有日志内容带error的日志。windowedCounts3 = words \.filter("content like '%error%'")# 开始运行查询并在控制台输出query = windowedCounts3 \.writeStream \.outputMode("update") \.format("console") \.option('truncate', 'false')\.trigger(processingTime="3 seconds") \.start()query.awaitTermination()

四、结果分析与实验体会

        Spark Structured Streaming 是 Spark 提供的用于实时流处理的 API，它提供了一种统一的编程模型，使得批处理和流处理可以共享相同的代码逻辑，让开发者更容易地实现复杂的实时流处理任务。通过对 Structured Streaming 的实验，有以下体会：

  1. 简单易用: Structured Streaming 提供了高级抽象的 DataFrame 和 Dataset API，使得流处理变得类似于静态数据处理，降低了学习成本和编程复杂度。

  2. 容错性强大: Structured Streaming 内置了端到端的 Exactly-Once 语义，能够保证在发生故障时数据处理的准确性，给开发者提供了更可靠的数据处理保障。

  3. 灵活性和扩展性: Structured Streaming 支持丰富的数据源和数据接收器，可以方便地与其他数据存储和处理系统集成，同时也支持自定义数据源和输出操作，满足各种不同场景的需求。

  4. 优化性能: Structured Streaming 内置了优化器和调度器，能够根据任务的特性自动优化执行计划，提升处理性能，同时还可以通过调整配置参数和优化代码来进一步提高性能。

  5. 监控和调试: Structured Streaming 提供了丰富的监控指标和集成的调试工具，帮助开发者实时监控作业运行状态、诊断问题，并进行性能调优。

        通过实验和实践，更深入地理解 Structured Streaming 的特性和工作原理，掌握实时流处理的开发技巧和最佳实践，为构建稳健可靠的实时流处理应用打下坚实基础。

        Syslog 是一种常用的日志标准，它定义了一个网络协议，用于在计算机系统和网络设备之间传递事件消息和警报。通过对 Syslog 的实验，有以下体会：

  1. 灵活性: Syslog 可以用于收集各种类型的事件和日志信息，包括系统日志、安全事件、应用程序消息等等，具有很高的灵活性和可扩展性。

  2. 可靠性: Syslog 提供了可靠的传输和存储机制，确保事件和日志信息不会丢失或损坏，在故障恢复和安全审计方面非常重要。

  3. 标准化: Syslog 是一种通用的日志标准，已经被广泛采用和支持，可以与各种操作系统、应用程序、设备和服务集成，提供了统一的数据格式和接口。

  4. 安全性: Syslog 支持基于 TLS 和 SSL 的加密和身份认证机制，确保传输的信息不会被窃听或篡改，保证了日志传输的安全性。

  5. 可视化: 通过将 Syslog 收集到集中式的日志管理系统中，可以方便地进行搜索、分析和可视化，使日志信息变得更加易于理解和利用。

        通过实验和实践，更深入地了解 Syslog 的工作原理和应用场景，学会如何配置和使用 Syslog，掌握日志收集、存储、分析和可视化的技巧和最佳实践，为构建高效、可靠、安全的日志管理系统打下坚实基础。

这篇关于Spark编程实验五：Spark Structured Streaming编程的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/703752。

相关文章

Linux 网络编程 --- 应用层

Linux 网络编程 --- 应用层

一、自定义协议和序列化反序列化 代码： 序列化反序列化实现网络版本计算器 二、HTTP协议 1、谈两个简单的预备知识 https://www.baidu.com/ --- 域名 --- 域名解析 --- IP地址 http的端口号为80端口，https的端口号为443 url为统一资源定位符。CSDNhttps://mp.csdn.net/mp_blog/creation/editor
阅读更多...
【Python编程】Linux创建虚拟环境并配置与notebook相连接

【Python编程】Linux创建虚拟环境并配置与notebook相连接

1.创建 使用 venv 创建虚拟环境。例如，在当前目录下创建一个名为 myenv 的虚拟环境： python3 -m venv myenv 2.激活 激活虚拟环境使其成为当前终端会话的活动环境。运行： source myenv/bin/activate 3.与notebook连接 在虚拟环境中，使用 pip 安装 Jupyter 和 ipykernel： pip instal
阅读更多...
【编程底层思考】垃圾收集机制，GC算法，垃圾收集器类型概述

【编程底层思考】垃圾收集机制，GC算法，垃圾收集器类型概述

Java的垃圾收集（Garbage Collection，GC）机制是Java语言的一大特色，它负责自动管理内存的回收，释放不再使用的对象所占用的内存。以下是对Java垃圾收集机制的详细介绍： 一、垃圾收集机制概述： 对象存活判断：垃圾收集器定期检查堆内存中的对象，判断哪些对象是“垃圾”，即不再被任何引用链直接或间接引用的对象。内存回收：将判断为垃圾的对象占用的内存进行回收，以便重新使用。
阅读更多...
Go Playground 在线编程环境

Go Playground 在线编程环境

For all examples in this and the next chapter, we will use Go Playground. Go Playground represents a web service that can run programs written in Go. It can be opened in a web browser using the follow
阅读更多...
深入理解RxJava：响应式编程的现代方式

深入理解RxJava：响应式编程的现代方式

在当今的软件开发世界中，异步编程和事件驱动的架构变得越来越重要。RxJava，作为响应式编程（Reactive Programming）的一个流行库，为Java和Android开发者提供了一种强大的方式来处理异步任务和事件流。本文将深入探讨RxJava的核心概念、优势以及如何在实际项目中应用它。 文章目录 💯 什么是RxJava？💯 响应式编程的优势💯 RxJava的核心概念
阅读更多...
函数式编程思想

函数式编程思想

我们经常会用到各种各样的编程思想，例如面向过程、面向对象。不过笔者在该博客简单介绍一下函数式编程思想. 如果对函数式编程思想进行概括，就是f(x) = na(x) , y=uf(x)…至于其他的编程思想，可能是y=a(x)+b(x)+c(x)…，也有可能是y=f(x)=f(x)/a + f(x)/b+f(x)/c… 面向过程的指令式编程 面向过程，简单理解就是y=a(x)+b(x)+c(x)
阅读更多...
STM32(十一)：ADC数模转换器实验

STM32(十一)：ADC数模转换器实验

AD单通道： 1.RCC开启GPIO和ADC时钟。配置ADCCLK分频器。 2.配置GPIO，把GPIO配置成模拟输入的模式。 3.配置多路开关，把左面通道接入到右面规则组列表里。 4.配置ADC转换器， 包括AD转换器和AD数据寄存器。单次转换，连续转换；扫描、非扫描；有几个通道，触发源是什么，数据对齐是左对齐还是右对齐。 5.ADC_CMD 开启ADC。 void RCC_AD
阅读更多...
Java并发编程之——BlockingQueue（队列）

Java并发编程之——BlockingQueue（队列）

一、什么是BlockingQueue BlockingQueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种： 1. 当队列满了的时候进行入队列操作2. 当队列空了的时候进行出队列操作123 因此，当一个线程试图对一个已经满了的队列进行入队列操作时，它将会被阻塞，除非有另一个线程做了出队列操作；同样，当一个线程试图对一个空
阅读更多...
Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering)

Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering)

Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering) Power Iteration Clustering (PIC) 是一种基于图的聚类算法，用于在大规模数据集上进行高效的社区检测。PIC 算法的核心思想是通过迭代图的幂运算来发现数据中的潜在簇。该算法适用于处理大规模图数据，特别是在社交网络分析、推荐系统和生物信息学等领域具有广泛应用。Spa
阅读更多...
生信代码入门：从零开始掌握生物信息学编程技能

生信代码入门：从零开始掌握生物信息学编程技能

少走弯路,高效分析;了解生信云,访问 【生信圆桌x生信专用云服务器】 : www.tebteb.cc 介绍 生物信息学是一个高度跨学科的领域，结合了生物学、计算机科学和统计学。随着高通量测序技术的发展，海量的生物数据需要通过编程来进行处理和分析。因此，掌握生信编程技能，成为每一个生物信息学研究者的必备能力。 生信代码入门，旨在帮助初学者从零开始学习生物信息学中的编程基础。通过学习常用
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2