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展示各个指标的信息。**

架构

数据采集：采用Flume NG进行数据采集
数据汇总和转发：用Flume 将数据转发和汇总到实时消息系统Kafka
数据处理：采用spark streming 进行实时的数据处理
结果显示：flask作为可视化工具进行结果显示

数据采集：通过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中

#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

代码

移步: github.com/jinhang