正在学习这篇文章：
http://blog.csdn.net/ymh198816/article/details/51998085
和工作中接触的电商、订单、分析，可以结合起来。
开宗明义，这幅图片：
Strom是一个非常快的实时计算框架，至于快到什么程度呢？
官网首页给出的数据是每一个Storm集群上的节点每一秒能处理一百万条数据。
相比Hadoop的“Mapreduce”计算框架，Storm使用的是"Topology"；
Mapreduce程序在计算完成后最终会停下来，而Topology则是会永远运行下去除非你显式地使用“kill -9 XXX”命令停掉它。
准备实际写一个实时分析系统。不然纸上得来终觉浅。
首先需要让Java程序在Linux环境上运行。用leetcode的java程序来做实验，把leetcode工程的output目录拷贝到安装了java的机器（m42n05.gzns）。
$ pwd /home/work/data/code/out/production/leetcode $ java com.company.Main Hello! ret:3 注意，只能在这个目录，运行完成的package。如果换到子目录，就不行： $ pwd /home/work/data/code/out/production/leetcode/com $ java company.Main Error: Could not find or load main class company.Main
用Idea创建了一个Maven项目“LogGenerator”，项目的主要代码如下：
package com.comany.log.generator; /** * Created by baidu on 16/11/7. */ import java.text.SimpleDateFormat; import java.util.Date; import java.util.Random; // Import log4j classes. import org.apache.log4j.LogManager; import org.apache.log4j.Logger; public class LogGenerator { public enum paymentWays { Wechat,Alipay,Paypal } public enum merchantNames { 优衣库,天猫,淘宝,咕噜大大,快乐宝贝,守望先峰,哈毒妇,Storm,Oracle,Java,CSDN,跑男,路易斯威登, 暴雪公司,Apple,Sumsam,Nissan,Benz,BMW,Maserati } public enum productNames { 黑色连衣裙, 灰色连衣裙, 棕色衬衫, 性感牛仔裤, 圆脚牛仔裤,塑身牛仔裤, 朋克卫衣,高腰阔腿休闲裤,人字拖鞋, 沙滩拖鞋 } float[] skuPriceGroup = {299,399,699,899,1000,2000}; float[] discountGroup = {10,20,50,100}; float totalPrice = 0; float discount = 0; float paymentPrice = 0; private static final Logger logger = LogManager.getLogger(LogGenerator.class); private int logsNumber = 10; public void generate() { for(int i = 0; i <= logsNumber; i++) { logger.info(randomOrderInfo()); } } public String randomOrderInfo() { SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); Date date = new Date(); String orderNumber = randomNumbers(5) + date.getTime(); String orderDate = sdf.format(date); String paymentNumber = randomPaymentWays() + "-" + randomNumbers(8); String paymentDate = sdf.format(date); String merchantName = randomMerchantNames(); String skuInfo = randomSkus(); String priceInfo = calculateOrderPrice(); return "orderNumber: " + orderNumber + " | orderDate: " + orderDate + " | paymentNumber: " + paymentNumber + " | paymentDate: " + paymentDate + " | merchantName: " + merchantName + " | sku: " + skuInfo + " | price: " + priceInfo; } private String randomPaymentWays() { paymentWays[] paymentWayGroup = paymentWays.values(); Random random = new Random(); return paymentWayGroup[random.nextInt(paymentWayGroup.length)].name(); } private String randomMerchantNames() { merchantNames[] merchantNameGroup = merchantNames.values(); Random random = new Random(); return merchantNameGroup[random.nextInt(merchantNameGroup.length)].name(); } private String randomProductNames() { productNames[] productNameGroup = productNames.values(); Random random = new Random(); return productNameGroup[random.nextInt(productNameGroup.length)].name(); } private String randomSkus() { Random random = new Random(); int skuCategoryNum = random.nextInt(3); String skuInfo ="["; totalPrice = 0; for(int i = 1; i <= 3; i++) { int skuNum = random.nextInt(3)+1; float skuPrice = skuPriceGroup[random.nextInt(skuPriceGroup.length)]; float totalSkuPrice = skuPrice * skuNum; String skuName = randomProductNames(); String skuCode = randomCharactersAndNumbers(10); skuInfo += " skuName: " + skuName + " skuNum: " + skuNum + " skuCode: " + skuCode + " skuPrice: " + skuPrice + " totalSkuPrice: " + totalSkuPrice + ";"; totalPrice += totalSkuPrice; } skuInfo += " ]"; return skuInfo; } private String calculateOrderPrice() { Random random = new Random(); discount = discountGroup[random.nextInt(discountGroup.length)]; paymentPrice = totalPrice - discount; String priceInfo = "[ totalPrice: " + totalPrice + " discount: " + discount + " paymentPrice: " + paymentPrice +" ]"; return priceInfo; } private String randomCharactersAndNumbers(int length) { String characters = "abcdefghijklmnopqrstuvwxyz0123456789"; String randomCharacters = ""; Random random = new Random(); for (int i = 0; i < length; i++) { randomCharacters += characters.charAt(random.nextInt(characters.length())); } return randomCharacters; } private String randomNumbers(int length) { String characters = "0123456789"; String randomNumbers = ""; Random random = new Random(); for (int i = 0; i < length; i++) { randomNumbers += characters.charAt(random.nextInt(characters.length())); } return randomNumbers; } public static void main(String[] args) { LogGenerator generator = new LogGenerator(); generator.generate(); } }
运行的时候报错：
log4j:WARN No appenders could be found for logger (com.comany.log.generator.LogGenerator). log4j:WARN Please initialize the log4j system properly. log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
所以增加一个 log4j.properties
log4j.rootLogger=INFO,Console,File #控制台日志 log4j.appender.Console=org.apache.log4j.ConsoleAppender log4j.appender.Console.Target=System.out log4j.appender.Console.layout=org.apache.log4j.PatternLayout log4j.appender.Console.layout.ConversionPattern=[%p][%t][%d{yyyy-MM-dd HH:mm:ss}][%C] - %m%n #普通文件日志 log4j.appender.File=org.apache.log4j.RollingFileAppender log4j.appender.File.File=logs/generator.log log4j.appender.File.MaxFileSize=10MB #输出日志，如果换成DEBUG表示输出DEBUG以上级别日志 log4j.appender.File.Threshold=ALL log4j.appender.File.layout=org.apache.log4j.PatternLayout log4j.appender.File.layout.ConversionPattern=[%p][%t][%d{yyyy-MM-dd HH:mm:ss}][%C] - %m%n
然后在外层目录能够看到有Log目录生成：
直接把target目录拷贝到Linux机器上，发现找不到依赖包 log4j。
需要用 Intellij 进行打包。在File -> Project Structure里面。
然后应该会自动生成Jar包（也可以Build->Build Artifacts） LogGenerator.jar ，拷贝到Linux机器上。
但是开始不能运行。提示找不到Manifest.mf。搜索之后，发现，要选第二个选项“copy to ...”，而且必须把Manifest的目录从java改到Resources才行。
这样选项之后，生成的目录里面有两个jar。目录拷贝到Linux，然后运行：
$ java -jar LogGenerator.jar 能够看到logs目录下有新的日志生成： $ ll logs/* -rw-rw-r-- 1 work work 12328 Nov 7 20:12 logs/generator.log
但是Linux的log文件都是乱码。试了改SecureCRT配置什么的，都没有用。
查看到文件编码是 ASCII TEXT
只好用 (printf "357273277";cat generator.log) > File2 来修改文件编码格式。
还是乱码。在log4j的配置文件里面加上
log4j.appender.file.encoding=UTF-8
还是乱码。加上上面哪个 BOM符号，还是乱码。
这时候把mac上面的日志拷贝到Linux上，发现是正常的。那么还是log4j打印的地方出了问题。
再仔细检查log4j的地方，发现上面log4j配置里面的file应该大写才行，要与上下文一致：
log4j.appender.File.encoding=UTF-8
修改之后，重新生成artifact，拷贝，运行。能看到中文啦：
[INFO][main][2016-11-07 23:50:51][com.comany.log.generator.LogGenerator] - orderNumber: 494341478533851064 | orderDate: 2016-11-07 23:50:51 | paymentNumber: Alipay-46983228 | paymentDate: 2016-11-07 23:50:51 | merchantName: 守望先峰 | sku: [ skuName: 黑色连衣裙 skuNum: 3 skuCode: 06vteu0ewx skuPrice: 899.0 totalSkuPrice: 2697.0; skuName: 塑身牛仔裤 skuNum: 3 skuCode: oz13bdht0w skuPrice: 2000.0 totalSkuPrice: 6000.0; skuName: 圆脚牛仔裤 skuNum: 3 skuCode: geuum757jk skuPrice: 399.0 totalSkuPrice: 1197.0; ] | price: [ totalPrice: 9894.0 discount: 100.0 paymentPrice: 9794.0 ]
[INFO][main][2016-11-07 23:50:51][com.comany.log.generator.LogGenerator] - orderNumber: 623011478533851071 | orderDate: 2016-11-07 23:50:51 | paymentNumber: Alipay-58335677 | paymentDate: 2016-11-07 23:50:51 | merchantName: 暴雪公司 | sku: [ skuName: 黑色连衣裙 skuNum: 1 skuCode: mp9fbajaj9 skuPrice: 699.0 totalSkuPrice: 699.0; skuName: 黑色连衣裙 skuNum: 1 skuCode: oaiwi0xj3z skuPrice: 2000.0 totalSkuPrice: 2000.0; skuName: 黑色连衣裙 skuNum: 1 skuCode: tkbd5a91iq skuPrice: 399.0 totalSkuPrice: 399.0; ] | price: [ totalPrice: 3098.0 discount: 100.0 paymentPrice: 2998.0 ]
安装Flume
参考 http://www.aboutyun.com/thread-8917-1-1.html
http://www.cnblogs.com/smartloli/p/4468708.html
下载了apache-flume-1.7.0-bin.tar.gz ， 拷贝到 m42n06:/home/work/data/installed
解压后在conf目录 cp flume-env.sh.template flume-env.sh
把JAVA_HOME改对：
export JAVA_HOME=/home/work/.jumbo/opt/sun-java8/
运行命令，报错：
$ bin/flume-ng version bin/flume-ng: line 82: syntax error in conditional expression: unexpected token `(' bin/flume-ng: line 82: syntax error near `^java.library.path=(.' bin/flume-ng: line 82: ` if [[ $line =~ ^java.library.path=(.*)$ ]]; then'
怀疑又是shell脚本的原因。。
用jumbo 装了一个 zsh，但是执行的时候貌似还是有点问题。再用jumbo装一个coreutils试试吧。
装好coreutils之后 bash版本没有变化，试了一下还是不行，只好还是用zsh。
把flume-ng脚本的第一行改成 #!/home/work/.jumbo/bin/zsh-5.0.2
运行报错：
$ bin/flume-ng version bin/flume-ng version bin/flume-ng:cd:409: too many arguments run_flume:13: no such file or directory: /home/work/data/installed/apache-flume-1.7.0-bin/bin/java
发现原来是要指定配置文件，而且不能指定文件名，只能指定目录，如下：
$ bin/flume-ng version -c conf bin/flume-ng version -c conf Flume 1.7.0 Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git Revision: 511d868555dd4d16e6ce4fedc72c2d1454546707 Compiled by bessbd on Wed Oct 12 20:51:10 CEST 2016 From source with checksum 0d21b3ffdc55a07e1d08875872c00523
准备配一个flume，sink输出到hdfs上面，注意我们能够通过两种命令方式访问hdfs上面的文件：
$ ../hadoop-2.7.3/bin/hdfs dfs -cat /output/part-00000 ../hadoop-2.7.3/bin/hdfs dfs -cat /output/part-00000 5 5 15 $ ../hadoop-2.7.3/bin/hadoop fs -cat /output/part-00000 ../hadoop-2.7.3/bin/hadoop fs -cat /output/part-00000 5 5 15
配置成hdfs的方式：
agent1.sources = origin agent1.channels = memoryChannel agent1.sinks = hsink # For each one of the sources, the type is defined agent1.sources.origin.type = exec agent1.sources.origin.command = tail -f /home/work/data/LogGenerator_jar/logs/generator.log # The channel can be defined as follows. agent1.sources.origin.channels = memoryChannel # Each sink's type must be defined agent1.sinks.hsink.type = hdfs agent1.sinks.hsink.hdfs.path = /output/logOut agent1.sinks.hsink.hdfs.fileType = DataStream agent1.sinks.hsink.hdfs.writeFormat=TEXT agent1.sinks.hsink.hdfs.rollInterval=1 agent1.sinks.hsink.hdfs.filePrefix=%Y-%m-%d #Specify the channel the sink should use agent1.sinks.hsink.channel = memoryChannel # Each channel's type is defined. agent1.channels.memoryChannel.type = memory # Other config values specific to each type of channel(sink or source) # can be defined as well # In this case, it specifies the capacity of the memory channel agent1.channels.memoryChannel.capacity = 100
然后运行命令，还是报错：
$ bin/flume-ng agent -n agent -f conf/flume-conf.properties -c conf Info: Sourcing environment configuration script /home/work/data/installed/apache-flume-1.7.0-bin/conf/flume-env.sh Info: Including Hive libraries found via () for Hive access +run_flume:13> /home/work/.jumbo/opt/sun-java8//bin/java -Xmx20m -cp '/home/work/data/installed/apache-flume-1.7.0-bin/conf:/home/work/data/installed/apache-flume-1.7.0-bin/lib/*:/lib/*' '-Djava.library.path=' org.apache.flume.node.Application ' -n agent1 -f flume-conf.properties'
日志 logs/flums.log显示：
08 Nov 2016 17:17:01,783 ERROR [main] (org.apache.flume.node.Application.main:348) - A fatal error occurred while running. Exception follows. org.apache.commons.cli.MissingOptionException: Missing required option: n at org.apache.commons.cli.Parser.checkRequiredOptions(Parser.java:299) at org.apache.commons.cli.Parser.parse(Parser.java:231) at org.apache.commons.cli.Parser.parse(Parser.java:85) at org.apache.flume.node.Application.main(Application.java:263)
无语。。
又安装了flume 1.5试了下，还是不行。
又仔细检查了一下，很可能是java版本太高导致的。只能再装下java1.7试试看。发现还是不行。
最后发现是运行的java命令里面貌似有点问题：
/home/work/.jumbo/opt/sun-java8//bin/java -Xmx20m -cp '/home/work/data/installed/apache-flume-1.7.0-bin/conf:/home/work/data/installed/apache-flume-1.5.0-bin/lib/*' '-Djava.library.path=' org.apache.flume.node.Application ' -n agent -f flume-conf.properties'
需要把最后的引号去掉。
然后改成向日志输出。配置文件flume-conf.properties如下：
agent.sources = origin agent.channels = memoryChannel agent.sinks = loggerSink # For each one of the sources, the type is defined agent.sources.origin.type = exec agent.sources.origin.command = tail -f /home/work/data/LogGenerator_jar/logs/generator.log # The channel can be defined as follows. agent.sources.origin.channels = memoryChannel # Each sink's type must be defined agent.sinks.loggerSink.type = logger #Specify the channel the sink should use agent.sinks.loggerSink.channel = memoryChannel # Each channel's type is defined. agent.channels.memoryChannel.type = memory # Other config values specific to each type of channel(sink or source) # can be defined as well # In this case, it specifies the capacity of the memory channel agent.channels.memoryChannel.capacity = 100
Java命令如下：
/home/work/.jumbo/opt/sun-java8//bin/java -Xmx20m -cp '/home/work/data/installed/apache-flume-1.7.0-bin/conf:/home/work/data/installed/apache-flume-1.5.0-bin/lib/*' '-Dflume.root.logger=INFO,console' org.apache.flume.node.Application -n agent -f conf/flume-conf.properties
然后貌似能够监听到日志了。。。
其他的地方，还要再处理。。。
不知道为什么。登出重新登录之后，居然就好了。。。
用了HDFS作为sink的新的配置文件内容:
agent.sources = origin agent.channels = memoryChannel agent.sinks = hdfsSink # For each one of the sources, the type is defined agent.sources.origin.type = exec agent.sources.origin.command = tail -f /home/work/data/LogGenerator_jar/logs/generator.log # The channel can be defined as follows. agent.sources.origin.channels = memoryChannel # Each sink's type must be defined agent.sinks.hdfsSink.type = hdfs agent.sinks.hdfsSink.hdfs.path = /output/Logger agent.sinks.hdfsSink.hdfs.fileType = DataStream agent.sinks.hdfsSink.hdfs.writeFormati = TEXT agent.sinks.hdfsSink.hdfs.rollInterval = 1 agent.sinks.hdfsSink.hdfs.filePrefix=%Y-%m-%d
# 不加下面这一行会一直报错：Expected timestamp in the Flume event headers
agent.sinks.hdfsSink.hdfs.useLocalTimeStamp = true #Specify the channel the sink should use agent.sinks.hdfsSink.channel = memoryChannel # Each channel's type is defined. agent.channels.memoryChannel.type = memory # Other config values specific to each type of channel(sink or source) # can be defined as well # In this case, it specifies the capacity of the memory channel agent.channels.memoryChannel.capacity = 100
然后运行命令：
bin/flume-ng agent -n agent -f conf/flume-conf.properties -c conf
然后查看日志 logs/flume.log，貌似成功了（因为直接tail -f 是有内容的）
08 Nov 2016 23:19:03,709 INFO [SinkRunner-PollingRunner-DefaultSinkProcessor] (org.apache.flume.sink.hdfs.BucketWriter.open:231) - Creating /output/Logger/2016-11-08.1478618341857.tmp 08 Nov 2016 23:19:04,818 INFO [hdfs-hdfsSink-roll-timer-0] (org.apache.flume.sink.hdfs.BucketWriter.close:357) - Closing /output/Logger/2016-11-08.1478618341857.tmp 08 Nov 2016 23:19:04,828 INFO [hdfs-hdfsSink-call-runner-9] (org.apache.flume.sink.hdfs.BucketWriter$8.call:618) - Renaming /output/Logger/2016-11-08.1478618341857.tmp to /output/Logger/2016-11-08.1478618341857 08 Nov 2016 23:19:04,835 INFO [hdfs-hdfsSink-roll-timer-0] (org.apache.flume.sink.hdfs.HDFSEventSink$1.run:382) - Writer callback called.
在另一台机器05上面看下HDFS是否生成。
$ ~/data/installed/hadoop-2.7.3/bin/hadoop fs -ls /output/Logger Found 5 items -rw-r--r-- 3 work supergroup 1187 2016-11-08 23:19 /output/Logger/2016-11-08.1478618341853 -rw-r--r-- 3 work supergroup 1182 2016-11-08 23:19 /output/Logger/2016-11-08.1478618341854 -rw-r--r-- 3 work supergroup 1188 2016-11-08 23:19 /output/Logger/2016-11-08.1478618341855 -rw-r--r-- 3 work supergroup 1196 2016-11-08 23:19 /output/Logger/2016-11-08.1478618341856 -rw-r--r-- 3 work supergroup 1176 2016-11-08 23:19 /output/Logger/2016-11-08.1478618341857 发现生成的是一个目录，然后里面有按照时间戳生成的文件
然后打开一个文件看看：
$ ~/data/installed/hadoop-2.7.3/bin/hadoop fs -cat /output/Logger/2016-11-08.1478618810929
[INFO][main][2016-11-08 23:26:50][com.comany.log.generator.LogGenerator] - orderNumber: 966811478618810487 | orderDate: 2016-11-08 23:26:50 | paymentNumber: Wechat-92539661 | paymentDate: 2016-11-08 23:26:50 | merchantName: Sumsam | sku: [ skuName: 人字拖鞋 skuNum: 1 skuCode: 1q9vaq3484 skuPrice: 2000.0 totalSkuPrice: 2000.0; skuName: 朋克卫衣 skuNum: 3 skuCode: 7vwna4abw2 skuPrice: 2000.0 totalSkuPrice: 6000.0; skuName: 塑身牛仔裤 skuNum: 3 skuCode: 50qem6vlid skuPrice: 699.0 totalSkuPrice: 2097.0; ] | price: [ totalPrice: 10097.0 discount: 100.0 paymentPrice: 9997.0 ]
采用HDFS输出成功。
然后安装 Kafka。
Kafka安装新起一篇：http://www.cnblogs.com/charlesblc/p/6046023.html