配置Nginx
- yum install nginx （在host99和host101）
- service nginx start开启服务
- ps -ef |grep nginx看一下进程
ps -ef |grep nginx root 28230 1 0 14:54 ? 00:00:00 nginx: master process /usr/sbin/nginx -c /etc/nginx/nginx.conf nginx 28231 28230 0 14:54 ? 00:00:00 nginx: worker process nginx 28232 28230 0 14:54 ? 00:00:00 nginx: worker process nginx 28234 28230 0 14:54 ? 00:00:00 nginx: worker process nginx 28235 28230 0 14:54 ? 00:00:00 nginx: worker process ....
- 在本机浏览器也可以通过ip + port(默认80端口)访问
HTML
- 很简单的网页配置，在/root/html下放了一个index.html文件
- 先test一下是否正常，在browser里面输入ip地址就可以看到了。
- 下一步配置一个输入框，模拟搜索数据。
- index.html是特别简单的代码，（我是前端渣- -
-
View Code
<!DOCTYPE html> <html> <body> <form action="/result.html" method="GET"> Please Input:<br> <input type="text" name="Input" value="Mickey"> <br> <br> <input type="submit" value="Submit"> </form> </body> </html>
- 跳转到的页面result.html就随便写一句话啦
log日志
- log日志放在/var/log/nginx目录的access.log中。
-
tail -f access.log在输入框输入数据，提交之后会看到log刷新，比如我输入“test”提交后log会刷新一条：
-
10.109.255.90 - - [11/May/2017:14:24:16 +0800] "GET /action_page.php?Input=test HTTP/1.1" 404 169 "http://10.3.242.101/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/602.4.8 (KHTML, like Gecko) Version/10.0.3 Safari/602.4.8"
配置Flume
- 安装：（在host101）
-
wget http://mirrors.tuna.tsinghua.edu.cn/apache/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz tar -xvf apache-flume-1.7.0-bin.tar.gz - 测试：
- 在conf文件夹下新建一个test.conf文件如下：
-
View Code
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 # Describe the sink a1.sinks.k1.type = logger # Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 - 然后，运行
-
bin/flume-ng agent -n a1 -c conf -f conf/test.conf - 然后在logs文件夹下运行如下，刷新log
-
tail -f flume.log - 然后telnet localhost 44444。随便输入一些数据。log刷新显示出
-
11 May 2017 15:26:55,372 INFO [SinkRunner-PollingRunner-DefaultSinkProcessor] (org.apache.flume.sink.LoggerSink.process:95) - Event: { headers:{} body: 64 61 6B 6A 64 63 0D dakjdc. } - 就没问题啦
Nginx(WebServer)端
- Nginx端配置source-side flume agent来收集访问日志流数据
-
WebAccLog.sources = NginxAccess WebAccLog.sinks = AvroSink WebAccLog.channels = MemChannel WebAccLog.sources.NginxAccess.type = exec WebAccLog.sources.NginxAccess.command = tail -f /var/log/nginx/access.log WebAccLog.sources.NginxAccess.batchSize = 10 WebAccLog.sources.NginxAccess.interceptors = itime WebAccLog.sources.NginxAccess.interceptors.itime.type = timestamp WebAccLog.sinks.AvroSink.type = avro WebAccLog.sinks.AvroSink.hostname = 10.3.242.99 WebAccLog.sinks.AvroSink.port = 4545 WebAccLog.channels.MemChannel.type = memory WebAccLog.sinks.AvroSink.channel = MemChannel WebAccLog.sources.NginxAccess.channels = MemChannel
- 注意到我们在这里配置对source配置了一下interceptor拦截器。flume中source采集的日志首先会传入ChannelProcessor，在其内首先会通过interceptors进行过滤加工，然后通过ChannelSelector选择channel。
- 这里配置了一个时间戳拦截器，后面会在指定hdfs.path的时候用到(根据时间戳来指定存放路径)。
- 关于拦截器：flume内部实现了很多拦截器，同时还是先虑InterceptorChain用来链式处理event。
- HostInterceptor：在所有拦截的events的header中上加上本机的host name或IP
- TimestampInterceptor：在所有拦截的events的header中上加上它处理该时间的时间(in millis)。
- 这里之前写的时候source忘记加s了，然后查一下log才发现的问题，说实话这个conf文件好容易眼花，要改的东西太多- -。
Hadoop/Spark cluster端
- Hadoop/Spark cluster端配置receiver- side flume agent来接收数据.
- 首先使用logger sink来测试下是否一切正常
- 启动cluster端的flume，然后在web server端运行
-
bin/flume-ng avro-client -c ./conf -H 10.3.242.99 -p 4545 -F /var/log/nginx/access.log - 然后cluster端的flume会在log里面打出一堆数据，说明可以了。
- 接下来，把两边连通
- 下面是一个HDFS sink的例子，其他需求可以通过修改sink部分实现。
-
# clusterLogAgent # Naming the components of the current agent. clusterLogAgent.sources = AvroSource clusterLogAgent.sinks = HDFS clusterLogAgent.channels = MemChannel # Source configuration clusterLogAgent.sources.AvroSource.type = avro # hostname or IP address to listen on clusterLogAgent.sources.AvroSource.bind = 0.0.0.0 clusterLogAgent.sources.AvroSource.port = 4545 # sink configuration(write to HDFS) clusterLogAgent.sinks.HDFS.type = hdfs clusterLogAgent.sinks.HDFS.hdfs.path = /logFlume/nginx/accesslog # File format: currently SequenceFile, DataStream or CompressedStream clusterLogAgent.sinks.HDFS.hdfs.fileType = DataStream # Number of events written to file before it rolled (0 = never roll based on number of events) clusterLogAgent.sinks.HDFS.hdfs.rollCount = 0 clusterLogAgent.channels.MemChannel.type = memory clusterLogAgent.sources.AvroSource.channels = MemChannel clusterLogAgent.sinks.HDFS.channel = MemChannel
启动
- 首先开启cluster上的flume服务(这里要注意开启顺序
-
./bin/flume-ng agent -n clusterLogAgent -c conf -f conf/flume-hdfsSink.conf
- 开启Nginx端的flume服务
-
bin/flume-ng agent -n WebAccLog -c conf -f conf/flume-avro.conf
- check HDFS来验证流访问log events是否成功写入
-
[root@host99 /home/hhh/apache-flume-1.7.0-bin/conf]$hadoop fs -ls /logFlume/nginx/accesslog 17/05/11 17:25:47 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable Found 3 items -rw-r--r-- 2 root supergroup 1146 2017-05-11 17:15 /logFlume/nginx/accesslog/FlumeData.1494494113448 -rw-r--r-- 2 root supergroup 1190 2017-05-11 17:15 /logFlume/nginx/accesslog/FlumeData.1494494113449 -rw-r--r-- 2 root supergroup 952 2017-05-11 17:16 /logFlume/nginx/accesslog/FlumeData.1494494113450
-
[root@host101 /home/hhh/apache-flume-1.7.0-bin/conf]$hadoop fs -cat /logFlume/nginx/accesslog/FlumeData.1494494113448 17/05/11 17:19:59 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable 10.109.255.90 - - [11/May/2017:16:56:35 +0800] "GET /result.html?Input=ttt HTTP/1.1" 200 23 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/602.4.8 (KHTML, like Gecko) Version/10.0.3 Safari/602.4.8" 10.109.255.90 - - [11/May/2017:16:56:45 +0800] "GET /result.html?Input=ttt HTTP/1.1" 200 23 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/602.4.8 (KHTML, like Gecko) Version/10.0.3 Safari/602.4.8" 10.109.255.90 - - [11/May/2017:16:57:04 +0800] "GET /result.html?Input=hhhhhh HTTP/1.1" 200 23 "http://10.3.242.101/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/602.4.8 (KHTML, like Gecko) Version/10.0.3 Safari/602.4.8" 10.109.255.90 - - [11/May/2017:16:57:07 +0800] "GET /result.html?Input=hhhhhh HTTP/1.1" 200 23 "http://10.3.242.101/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/602.4.8 (KHTML, like Gecko) Version/10.0.3 Safari/602.4.8"
Flume conf调优
可优化的配置：
- Avro Sink：
- compression-type：压缩类型，默认不压缩。
- compression-level: 压缩等级
- maxIoWorkers：最大的I/O worker线程个数，默认2*机器最大可用处理器数目
-
-
threads：最大worker线程数
-
-
HDFS sink：
-
HDFS sink支持创建text和sequence文件；支持压缩；
-
文件可以被rolled。rolled基于elapse time或数据size或events数量
- 可以根据events的属性(比如时间戳或机器)来划分数据(buckets/partitions data)
- HDFS路径可以包含格式化转移序列(formatiting escape sequences)，HDFS可以据其产生一个目录/文件名来存储events。如下：
-
Alias Description %{host} Substitute value of event header named “host”. Arbitrary header names are supported. %t Unix time in milliseconds %a locale’s short weekday name (Mon, Tue, ...) %A locale’s full weekday name (Monday, Tuesday, ...) %b locale’s short month name (Jan, Feb, ...) %B locale’s long month name (January, February, ...) %c locale’s date and time (Thu Mar 3 23:05:25 2005) %d day of month (01) %e day of month without padding (1) %D date; same as %m/%d/%y %H hour (00..23) %I hour (01..12) %j day of year (001..366) %k hour ( 0..23) %m month (01..12) %n month without padding (1..12) %M minute (00..59) %p locale’s equivalent of am or pm %s seconds since 1970-01-01 00:00:00 UTC %S second (00..59) %y last two digits of year (00..99) %Y year (2010) %z +hhmm numeric timezone (for example, -0400) %[localhost] Substitute the hostname of the host where the agent is running %[IP] Substitute the IP address of the host where the agent is running %[FQDN] Substitute the canonical hostname of the host where the agent is running -
- 一些可能的优化配置：
- hdfs.rollSize:触发roll的文件大小，in bytes，默认是1024
- hdfs.rollCount: 在roll之前写入文件的events数目，默认是10。设置为0的话就不会根据events数量roll了。
- hdfs.rollInterval:roll当前文件所等待的秒数，默认是9=30。设置为0的话就不会根据时间间隔roll了。
- hdfs.batchSize: 在flushed到HDFS之前写入到文件的events数目
- hdfs.threadsPoolSize:每个HDFS sink中HDFS IO操作的线程数目，默认10
- 一些可能的优化配置：
已完成的优化
- hdfs文件目录：按日期分目录
-
# flume-hdfs.sink
clusterLogAgent.sinks.HDFS.hdfs.path = /logFlume/nginx/%y.%m.%d/
然后就变成啦酱紫：
- 配置滚动文件的大小，避免产生一堆小文件
-
#clusterLogAgent.sinks.HDFS.hdfs.rollSize =64*1024*1024
clusterLogAgent.sinks.HDFS.hdfs.rollSize = 67108864
clusterLogAgent.sinks.HDFS.hdfs.rollCount = 0
clusterLogAgent.sinks.HDFS.hdfs.rollInterval = 0
TBD
- 压缩
- 线程、worker数目
- ...
- flume内部是可以做一些过滤的。基于当前场景的话，是要用到正则的，主要是考虑这样的话会不会很影响效率。因为flume内部也不适合做复杂的过滤。
Log Analyse
搜索词分析-Ngram
- 提取搜索词：正则。
-
10.109.255.90 - - [12/May/2017:14:31:46 +0800] "GET /result.html?Input=Mickey HTTP/1.1" 304 0 "http://10.3.242.101/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/602.4.8 (KHTML, like Gecko) Version/10.0.3 Safari/602.4.8" 10.30.146.74 - - [12/May/2017:14:33:55 +0800] "GET / HTTP/1.1" 304 0 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36"
可以看到就是要提取 "?Input=Mickey"这一部分。用正则“?Input=[^s]*s”来匹配这段。
还有一点就是看到有的log是没有input，状态是304的。这种请求是因为用了浏览器缓存。考虑到搜索结果不能是固定的，所以在conf里面加了一句add_header Cache-Control no-store;来禁用缓存。
- 在mapper的中建Pattern，而不是在map()函数中每次初始化。
- 用搜索词做Ngram，这里直接假设搜索词已经是'+'-split(在搜索框space分隔的数据会在log里变成+分隔的)。【TODO：后续可以实现切分】
- MapReduce实现：
- Job1: SplitNgram --> 正则匹配搜索词，切分搜索词，统计ngram出现次数；
- Job2： GetNgram --> 过滤count小于指定threshold的ngram，同时得到每个origin词联想到的topK。
模拟搜索
- nili小可爱真是可怜又坚强 什么也没有 零数据支持
- 模拟的方式简单，就写一个程序(python or java or scala or whatever)，就好啦
- 但是要稍微有点意义的搜索词，目前想到的方法就是找一堆数据，脚本一边读一边向nginx请求。
- 模拟http请求的demo如下：
-
import httplib coon = httplib.HTTPConnection('host101') word = 'test' conn.request('GET', 'result.html?input=' + word)
-
- 以及搜索数据目前能找到的方式就是，我在kaggle上看到的一个quora的比赛，可以拿到一些quora的问题，这在某种程度上也是搜索数据了吧。hhh自我满足ing...[所以其实后续想再扩展的话可以考虑爬quora呀23333] 加上搜索数据的整个模拟demo如下：【注：连接的效率等问题暂未考虑..
-
def get_search_word(file_name): # id, qid1, qid2, question1, question2, is_duplicate with open(file_name) as fi: for line in fi: splited = line.split(",") if len(splited) < 6: continue conn = httplib.HTTPConnection('10.3.242.101') conn.request('GET', 'result.html?input=' + splited[3]) # time.sleep(5) conn = httplib.HTTPConnection('10.3.242.101') conn.request('GET', 'result.html?input=' + splited[4]) time.sleep(5)