日志分析
单机日志分析，适用于小数据量的。（最大10G），awk/grep/sort/join等都是日志分析的利器。
例子：
1、shell得到Nginx日志中访问量最高的前十个IP
cat access.log.10 | awk '(a[$1]++) END (for(b in a) print b"\t"a[b])' | sort -k2 -r | head -n 102、python 统计每个IP的地址点击数
import re
import sys
contents=sys.argv[1]
def NginxIpHit(logfile_path):
ipadd = r'\.'.join([r'\d{1,3}']*4)
re_ip = re.compile(ipadd)
iphitlisting = {}
for line in open(contents):
match = re_ip.match(line)
if match:
ip = match.group()
iphitlisting[ip]=iphitlisting.get(ip,0)+1
print iphitlisting
NginxIpHit(contents)**大规模的日志处理，日志分析指标：
PV、UV、PUPV、漏斗模型和准化率、留存率、用户属性
最终用UI展示各个指标的信息。**
架构
- 1、实时日志处理流线
数据采集：采用Flume NG进行数据采集
数据汇总和转发：用Flume 将数据转发和汇总到实时消息系统Kafka
数据处理：采用spark streming 进行实时的数据处理
结果显示：flask作为可视化工具进行结果显示
- 2、离线日志处理流线
数据采集：通过Flume将数据转存到HDFS
数据处理：使用spark sql进行数据的预处理
结果呈现：结果汇总到mysql上，最后使用flask进行结果的展现
Lamda架构：低响应延迟的组合数据传输环境。
查询过程：一次流处理、一次批处理。对应着实时和离线处理。
项目流程
安装flume
Flume进行日志采集，web端的日志一般Nginx、IIS、Tomcat等。Tomcat的日志在var/data/log
安装jdk
安装Flume
wget http://mirrors.cnnic.cn/apache/flume/1.5.0/apache-flume-1.5.0-bin.tar.gz
tar –zxvf apache-flume-1.5.0-bin.tar.gz
mv apache-flume-1.5.0 –bin apache-flume-1.5.0
ln -s apache-flume-1.5.0 fiume 环境变量配置
Vim /etc/profile
Export JAVA_HOME=/usr/local/jdk
Export CLASS_PATH = .:$ JAVA_HOME/lib/dt.jar: $ JAVA_HOME/lib/tools.jar
Export PATH=$ PATH:$ JAVA_HOME/bin
Export FlUME_HOME=/usr/local/flume
Export FlUME_CONF_DIR=$ FlUME_HOME/conf
Export PATH=$ PATH:$ FlUME_HOME /bin
Souce /etc/profile 创建agent配置文件将数据输出到hdfs上，修改flume.conf:
a1.sources = r1
a1.sinks = k1
a1.channels =c1
描述和配置sources
第一步：配置数据源
a1.sources.r1.type =exec
a1.sources.r1.channels =c1
配置需要监控的日志输出目录
a1.sources.r1.command=tail –f /va/log/data
第二步：配置数据输出
a1.sink.k1.type =hdfs
a1.sink.k1.channels =c1
a1.sink.k1.hdfs.useLocalTimeStamp=true
a1.sink.k1.hdfs.path =hdfs://192.168.11.177:9000/flume/events/%Y/%m/%d/%H/%M
a1.sink.k1.hdfs.filePrefix =cmcc
a1.sink.k1.hdfs.minBlockReplicas=1
a1.sink.k1.hdfs.fileType =DataStream
a1.sink.k1.hdfs.writeFormat=Text
a1.sink.k1.hdfs.rollInterval =60
a1.sink.k1.hdfs.rollSize =0
a1.sink.k1.hdfs.rollCount=0
a1.sink.k1.hdfs.idleTimeout =0
配置数据通道
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
第四步：将三者级联
a1.souces.r1.channels =c1
a1.sinks.k1.channel =c1启动Flume Agent
cd /usr/local/flume
nohup bin/flume-ng agent –n conf -f conf/flume-conf.properties
&已经将flume整合到了hdfs中
- 整合Flume、kafka、hhdfs
#hdfs输出端
a1.sink.k1.type =hdfs
a1.sink.k1.channels =c1
a1.sink.k1.hdfs.useLocalTimeStamp=true
a1.sink.k1.hdfs.path =hdfs://192.168.11.174:9000/flume/events/%Y/%m/%d/%H/%M
a1.sink.k1.hdfs.filePrefix =cmcc-%H
a1.sink.k1.hdfs.minBlockReplicas=1
a1.sink.k1.hdfs.fileType =DataStream
a1.sink.k1.hdfs.rollInterval =3600
a1.sink.k1.hdfs.rollSize =0
a1.sink.k1.hdfs.rollCount=0
a1.sink.k1.hdfs.idleTimeout =0
#kafka输出端 为了提高性能使用内存通道
a1.sink.k2.type =com.cmcc.chiwei.Kafka.CmccKafkaSink
a1.sink.k2.channels =c2
a1.sink.k2.metadata.broker.List=192.168.11.174:9002;192.168.11.175:9092; 192.168.11.174:9092
a1.sink.k2.partion.key =0
a1.sink.k2.partioner.class= com.cmcc.chiwei.Kafka.Cmcc Partion
a1.sink.k2.serializer.class= kafka. Serializer.StringEncoder
a1.sink.k2.request.acks=0
a1.sink.k2.cmcc.encoding=UTF-8
a1.sink.k2.cmcc.topic.name=cmcc
a1.sink.k2.producer.type =async
a1.sink.k2.batchSize =100
a1.sources.r1.selector.type=replicating
a1.sources = r1
a1.sinks = k1 k2
a1.channels =c1 c2
#c1
a1.channels.c1.type=file
a1.channels.c1.checkpointDir=/home/flume/flumeCheckpoint
a1.channels.c1.dataDir=/home/flume/flumeData, /home/flume/flumeDataExt
a1.channels.c1.capacity=2000000
a1.channels.c1.transactionCapacity=100
#c2
a1.channels.c2.type=memory
a1.channels.c2.capacity=2000000
a1.channels.c2.transactionCapacity=100
用Kafka将日志汇总
1.4 Tar –zxvf kafka_2.10-0.8.1.1.tgz
1.5 配置kafka和zookeeper文件
配置zookeeper.properties
dataDir=/tmp/zookeeper
client.Port=2181
maxClientCnxns = 0
initLimit = 5
syncLimit = 2
##
server.43 = 10.190.182.43:2888:3888
server.38 = 10.190.182.38:2888:3888
server.33 = 10.190.182.33:2888:3888
配置zookeeper myid
在每个服务器dataDir 创建 myid文件 写入本机id
//server.43 myid 本机编号43
echo “43” > /tmp/ zookeeper/myid
配置kafka文件, config/server.properties
每个节点根据不同主机名配置
broker.id ：43
host.name:10.190.172.43
zookeeper.connect=10.190.172.43:2181, 10.190.172.33:2181，10.190.172.38:2181启动zookeeper
kafka通过zookeeper存储元数据，先启动它，提供kafka相应的连接地址
Kafka自带的zookeeper
在每个节点 bin/zookeeper-server-start.sh config/zookeeper. properties
启动Kafka
Bin/Kafka-server-start.sh创建和查看topic
Topic和flume中的要一致，spark streming 也用的这个
Bin/ Kafka-topics.sh --create --zookeeper 10.190.172.43:2181
--replication-factor 1 -- partions 1 --topic KafkaTopic查看下：
Bin/ Kafka-topics.sh --describe -- zookeeper 10.190.172.43:2181整合kafka sparkstreming
Buid.sbt
Spark-core
Spark-streming
Spark-streamng-kafka
kafka- Spark streming 实时分析
数据收集和中转已经好了，kafka给sparkstreming - Spark sql 离线分析
- Flask可视化
代码
移步: github.com/jinhang
时间： 2024-12-03 08:57:27