问题引入
- 数据生产者的分组策略？
- 如何保证数据的完全生产？
- partition数量和broker数量关系？
- 每个partition的数据如何保存到硬盘？
- Kafka有什么独特的特点？
- 消费者如何标记消费状态？
- 消费者负载均衡的策略？
- 如何保证消费者消费的数据是有序的？
生产者是集群模式，全局序号管理器。
broker端只设置一个partition，kafka的高并发下的负载均衡。
消费者如果是一个组，如何保证消息有序？消费者来一个线程(自定义一个数据结构来排序)
模拟应用需求
- 采集订单系统应用打印的日志文件。
日志文件使用log4j生成，滚动生成。使用tail -F xxx.log来监控文件名称，理解tail -f和tail -F的区别。
- 将采集的日志文件保存到Kafka中。
(source)输入：tail -F xxx.log
(channel)存储：内存
(sink)输出：Kafka
config样例，
a1.source = s1
a1.channel = c1
a1.sink = k1
source exec tail -F xxx.log
channel RAM
sink xxxx.xxxx.xxxx.KafkaSink // 该类必须存放lib目录
sink.topic = orderMq
sink.itcast = itcast
map = getConfig();
value = map.get("itcast")
- 通过Storm程序消费Kafka中数据。
KafkaSpout
Bolt1()
Bolt2()
业务模拟：统计双十一当前的订单金额、订单数量、订单人数。订单金额/数量/人数(整个网站、各个业务线、各个品类、各个店铺、各个品牌、每个商品)。
环境配置
应用安装的一般流程：下载、解压、配置、分发。
在Flume官方网站下载Flume，解压Flume安装包，
tar -zxvf apache-flume-1.6.0-bin.tar.gz -C /export/servers/
cd /export/servers/
ln -s apache-flume-1.6.0-bin flume
配置Flume环境变量，
vi /etc/profile
export FLUME_HOME=/export/servers/apache-flume-1.6.0-bin
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$ZK_HOME/bin:$STORM_HOME/bin:$KAFKA_HOME/bin:$FLUME_HOME/bin
source /etc/profile
创建Flume配置文件，
cd /export/servers/flume/conf/
mkdir myconf
cd myconf/
vi exec.conf
a1.sources = r1
a1.channels = c1
a1.sinks = k1
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/data/flume_sources/click_log/1.log
a1.sources.r1.channels = c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = orderMq
a1.sinks.k1.brokerList = kafka01:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20
a1.sinks.k1.channel = c1
准备模拟应用日志数据的目录，
mkdir -p /export/data/flume_sources/click_log
通过脚本模拟生产应用日志数据，
# click_log_out.sh
for((i=0;i<=50000;i++));
do echo "message-"+$i >> /export/data/flume_sources/click_log/1.log;
done
chomd +x click_log_out.sh
打通所有流程，第一步，启动zk集群，
zkServer.sh start
zkServer.sh status
第二步，封装Kafka集群启动和停止的脚本，启动Kafka集群，
start-kafka.sh
第三步，启动Flume客户端，监控日志数据生成，
./bin/flume-ng agent -n a1 -c conf -f conf/myconf/exec.conf -Dflume.root.logger=INFO,console
第四步，创建一个topic并开启consumer，在客户端模拟消费，
kafka-console-consumer.sh --zookeeper zk01:2181 --topic orderMq
第五步，执行应用日志数据生产脚本，
sh click_log_out.sh
整合Storm程序的bug解决
服务端没有启动ZooKeeper，
ERROR org.apache.curator.ConnectionState - Connection timed out for connection string (zk01:2181,zk02:2181,zk03:2181) and timeout (15000) / elapsed (15071)
本地调试Storm程序，本机没有配置kafka的解析，
kafka.consumer.SimpleConsumer - Reconnect due to error:
java.nio.channels.ClosedChannelException: null
放到Storm集群运行，相关环境和jar包冲突，把Storm相关的依赖去掉。
SLF4J: Actual binding is of type [ch.qos.logback.classic.util.ContextSelectorStaticBinder]
Exception in thread "main" java.lang.ExceptionInInitializerError
at backtype.storm.topology.TopologyBuilder.createTopology(TopologyBuilder.java:106)
at cn.itcast.storm.kafkaAndStorm.KafkaAndStormTopologyMain.main(KafkaAndStormTopologyMain.java:27)
Caused by: java.lang.RuntimeException: Found multiple defaults.yaml resources. You're probably bundling the Storm jars with your topology jar. [jar:file:/home/hadoop/kafka2storm.jar!/defaults.yaml, jar:file:/export/servers/apache-storm-0.9.5/lib/storm-core-0.9.5.jar!/defaults.yaml]
at backtype.storm.utils.Utils.findAndReadConfigFile(Utils.java:133)
at backtype.storm.utils.Utils.readDefaultConfig(Utils.java:160)
at backtype.storm.utils.Utils.readStormConfig(Utils.java:184)
at backtype.storm.utils.Utils.<clinit>(Utils.ja
本地模式，提供storm的核心jar包，
java.lang.NoClassDefFoundError: backtype/storm/topology/IRichSpout
at java.lang.Class.getDeclaredMethods0(Native Method)
at java.lang.Class.privateGetDeclaredMethods(Class.java:2615)
at java.lang.Class.getMethod0(Class.java:2856)
at java.lang.Class.getMethod(Class.java:1668)
at sun.launcher.LauncherHelper.getMainMethod(LauncherHelper.java:494)
at sun.launcher.LauncherHelper.checkAndLoadMain(LauncherHelper.java:486)
Caused by: java.lang.ClassNotFoundException: backtype.storm.topology.IRichSpout
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
... 6 more
Exception in thread "main"
Process finished with exit code 1
本文首发于steem，感谢阅读，转载请注明。
微信公众号「数据分析」，分享数据科学家的自我修养，既然遇见，不如一起成长。
数据分析
读者交流电报群
知识星球交流群
知识星球读者交流群
