Spark-再接着上次的Lamda架构

日志分析

单机日志分析，适用于小数据量的。（最大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 10

2、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