Structured Streaming编程练习-日志分析

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)

对Syslog进行查询由Spark接收nc程序发送过来的日志信息,然后完成以下任务:

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

ps:基于pyspark

新建python脚本test.py如下:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from functools import partial

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

if __name__ == "__main__":
    spark = SparkSession.builder.appName("syslog_ouput").getOrCreate()
    spark.sparkContext.setLogLevel('WARN')
    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('2020 %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'))

    # (2).  统计每小时的每个进程或者服务分别产生的日志总数,水印设置为1分钟。
    windowedCounts2 = words.withWatermark("timestamp", "1 minutes") .groupBy('tag', window('timestamp', "1 hour")).count().sort(asc('window'))

    #(3).   输出所有日志内容带error的日志
    #如果没有error信息,可以手动加入error日志信息,在终端输入下面语句即可
    #logger 'I am a test error log message.'
    windowedCounts3 = words.filter("content like '%error%'")

    # 开始运行查询并在控制台输出,问题3使用update输出,complete输出要有聚合函数
    query = windowedCounts1.writeStream.outputMode("complete").format("console").option('truncate', 'false').trigger(processingTime="3 seconds").start()
    #query = windowedCounts2.writeStream.outputMode("complete").format("console").option('truncate', 'false').trigger(processingTime="3 seconds").start()
    #query = windowedCounts3.writeStream.outputMode("update").format("console").option('truncate', 'false').trigger(processingTime="3 seconds").start()

    query.awaitTermination()

打开一个终端,输入下面语句:

向9988端口发送日志信息

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

再打开一个终端,运行Python脚本文件:

/usr/local/spark/bin/spark-submit test.py

对每小时CRON进程的信息计数后输出

Structured Streaming编程练习-日志分析

学习交流,有任何问题还请随时评论指出交流。

上一篇:蜂巢APP对接ELK统一日志


下一篇:搬家第24天-citect V7.4 CitectVBA使用ADO连接ACCESS数据库,并且把表内容填入microsoft flexgrid控件里面