Hadoop集群日志收集与分析

Hadoop集群日志收集与分析实战方案

背景与需求

在Hadoop生态环境中，集群日志收集与分析是运维监控的核心环节。本文提供一套完整的日志处理方案，涵盖日志采集、存储、分析全流程。

方案架构

[日志源] → [Flume/Kafka] → [HDFS] → [Spark/Hive] → [可视化]

1. 日志采集配置

使用Flume作为日志采集工具，配置文件如下：

# agent配置
agent.sources = r1
agent.sinks = k1
agent.channels = c1

# source配置
agent.sources.r1.type = spooldir
agent.sources.r1.spoolDir = /var/log/hadoop
agent.sources.r1.channels = c1

# sink配置
agent.sinks.k1.type = hdfs
agent.sinks.k1.hdfs.path = hdfs://namenode:9000/logs/%Y-%m-%d
agent.sinks.k1.hdfs.filePrefix = hadoop-log
agent.sinks.k1.hdfs.rollInterval = 3600
agent.sinks.k1.channels = c1

2. HDFS存储优化

创建日志目录结构并设置权限：

# 创建日志目录
hdfs dfs -mkdir -p /logs/hadoop/daily
hdfs dfs -chmod 755 /logs/hadoop

# 设置日志轮转策略
hdfs dfs -setrep 3 /logs/hadoop

3. 日志分析脚本

使用Spark SQL进行日志分析：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HadoopLogAnalysis") \
    .enableHiveSupport() \
    .getOrCreate()

# 读取日志数据
log_df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("hdfs://namenode:9000/logs/hadoop/*.csv")

# 统计错误日志
error_logs = log_df.filter(log_df.level == "ERROR")
error_logs.groupBy("component").count().show()

# 时间序列分析
log_df.createOrReplaceTempView("hadoop_logs")
spark.sql("""
    SELECT 
        date_format(timestamp, 'yyyy-MM-dd') as log_date,
        component,
        count(*) as error_count
    FROM hadoop_logs 
    WHERE level = 'ERROR'
    GROUP BY log_date, component
    ORDER BY log_date
""").show()

4. 实施步骤

1. 部署Flume agent并测试日志采集
2. 在HDFS上创建相应目录结构
3. 编写Spark分析脚本并提交执行
4. 设置定时任务自动执行分析

5. 监控告警

配置Grafana + Prometheus监控体系，实现异常日志的实时告警。

该方案已在多个生产环境中验证，可有效提升Hadoop集群运维效率。