devbox
很楠不爱3
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

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

作者：很楠不爱3 | 2024-02-20 06:08:28

赞

踩

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通常被记录在本地文件内，也可以被发送给远程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通常被记录在本地文件内，也可以被发送给远程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进行拆分成结构化字段，以下是示例代码：

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

        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命令，否则多次运行测试可能导致没有新数据生成。

  1. #!/usr/bin/env python3
  2.  
  3. from functools import partial
  4.  
  5. from pyspark.sql import SparkSession
  6. from pyspark.sql.functions import *
  7.  
  8.  
  9. if __name__ == "__main__":
  10.     spark = SparkSession \
  11.         .builder \
  12.         .appName("StructuredSyslog") \
  13.         .getOrCreate()
  14.  
  15.     lines = spark \
  16.         .readStream \
  17.         .format("socket") \
  18.         .option("host", "localhost") \
  19.         .option("port", 9988) \
  20.         .load()
  21.  
  22.     # Nov 24 13:17:01 spark CRON[18455]: (root) CMD (   cd / && run-parts --report /etc/cron.hourly)
  23.     # 定义一个偏应用函数，从固定的pattern获取日志内匹配的字段
  24.     fields = partial(
  25.         regexp_extract, str="value", pattern="^(\w{3}\s*\d{1,2} \d{2}:\d{2}:\d{2}) (.*?) (.*?)\[*\d*\]*: (.*)$"
  26.     )
  27.  
  28.     words = lines.select(
  29.         to_timestamp(format_string('2019 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),
  30.         fields(idx=2).alias("hostname"),
  31.         fields(idx=3).alias("tag"),
  32.         fields(idx=4).alias("content"),
  33.     )
  34.  
  35.     # (1).  统计CRON这个进程每小时生成的日志数，并以时间顺序排列，水印设置为1分钟。
  36.     windowedCounts1 = words \
  37.         .filter("tag = 'CRON'") \
  38.         .withWatermark("timestamp", "1 minutes") \
  39.         .groupBy(window('timestamp', "1 hour")) \
  40.         .count() \
  41.         .sort(asc('window'))
  42.  
  43.     # 开始运行查询并在控制台输出
  44.     query = windowedCounts1 \
  45.         .writeStream \
  46.         .outputMode("complete") \
  47.         .format("console") \
  48.         .option('truncate', 'false')\
  49.         .trigger(processingTime="3 seconds") \
  50.         .start()
  51.  
  52.     query.awaitTermination()

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

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

  1. #!/usr/bin/env python3
  2.  
  3. from functools import partial
  4.  
  5. from pyspark.sql import SparkSession
  6. from pyspark.sql.functions import *
  7.  
  8.  
  9. if __name__ == "__main__":
  10.     spark = SparkSession \
  11.         .builder \
  12.         .appName("StructuredSyslog") \
  13.         .getOrCreate()
  14.  
  15.     lines = spark \
  16.         .readStream \
  17.         .format("socket") \
  18.         .option("host", "localhost") \
  19.         .option("port", 9988) \
  20.         .load()
  21.  
  22.     # Nov 24 13:17:01 spark CRON[18455]: (root) CMD (   cd / && run-parts --report /etc/cron.hourly)
  23.     # 定义一个偏应用函数，从固定的pattern获取日志内匹配的字段
  24.     fields = partial(
  25.         regexp_extract, str="value", pattern="^(\w{3}\s*\d{1,2} \d{2}:\d{2}:\d{2}) (.*?) (.*?)\[*\d*\]*: (.*)$"
  26.     )
  27.  
  28.     words = lines.select(
  29.         to_timestamp(format_string('2019 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),
  30.         fields(idx=2).alias("hostname"),
  31.         fields(idx=3).alias("tag"),
  32.         fields(idx=4).alias("content"),
  33.     )
  34.  
  35.     # (2).  统计每小时的每个进程或者服务分别产生的日志总数，水印设置为1分钟。
  36.     windowedCounts2 = words \
  37.         .withWatermark("timestamp", "1 minutes") \
  38.         .groupBy('tag', window('timestamp', "1 hour")) \
  39.         .count() \
  40.         .sort(asc('window'))
  41.  
  42.     # 开始运行查询并在控制台输出
  43.     query = windowedCounts2 \
  44.         .writeStream \
  45.         .outputMode("complete") \
  46.         .format("console") \
  47.         .option('truncate', 'false')\
  48.         .trigger(processingTime="3 seconds") \
  49.         .start()
  50.  
  51.     query.awaitTermination()

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

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

  1. #!/usr/bin/env python3
  2.  
  3. from functools import partial
  4.  
  5. from pyspark.sql import SparkSession
  6. from pyspark.sql.functions import *
  7.  
  8.  
  9. if __name__ == "__main__":
  10.     spark = SparkSession \
  11.         .builder \
  12.         .appName("StructuredSyslog") \
  13.         .getOrCreate()
  14.  
  15.     lines = spark \
  16.         .readStream \
  17.         .format("socket") \
  18.         .option("host", "localhost") \
  19.         .option("port", 9988) \
  20.         .load()
  21.  
  22.     # Nov 24 13:17:01 spark CRON[18455]: (root) CMD (   cd / && run-parts --report /etc/cron.hourly)
  23.     # 定义一个偏应用函数，从固定的pattern获取日志内匹配的字段
  24.     fields = partial(
  25.         regexp_extract, str="value", pattern="^(\w{3}\s*\d{1,2} \d{2}:\d{2}:\d{2}) (.*?) (.*?)\[*\d*\]*: (.*)$"
  26.     )
  27.  
  28.     words = lines.select(
  29.         to_timestamp(format_string('2019 %s', fields(idx=1)), 'yy MMM d H:m:s').alias("timestamp"),
  30.         fields(idx=2).alias("hostname"),
  31.         fields(idx=3).alias("tag"),
  32.         fields(idx=4).alias("content"),
  33.     )
  34.  
  35.     # (3).  输出所有日志内容带error的日志。
  36.     windowedCounts3 = words \
  37.         .filter("content like '%error%'")
  38.  
  39.     # 开始运行查询并在控制台输出
  40.     query = windowedCounts3 \
  41.         .writeStream \
  42.         .outputMode("update") \
  43.         .format("console") \
  44.         .option('truncate', 'false')\
  45.         .trigger(processingTime="3 seconds") \
  46.         .start()
  47.  
  48.     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，掌握日志收集、存储、分析和可视化的技巧和最佳实践，为构建高效、可靠、安全的日志管理系统打下坚实基础。

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号