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进程的信息计数后输出
学习交流,有任何问题还请随时评论指出交流。