经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Apache Kafka » 查看文章
Flume+Kafka+Storm+Hbase+HDSF+Poi整合
来源：cnblogs　　作者：Hongten　　时间：2018/12/24 10:26:48　　对本文有异议

Flume+Kafka+Storm+Hbase+HDSF+Poi整合

需求：

针对一个网站，我们需要根据用户的行为记录日志信息，分析对我们有用的数据。

举例：这个网站www.hongten.com（当然这是一个我虚拟的电商网站），用户在这个网站里面可以有很多行为，比如注册，登录，查看，点击，双击，购买东西，加入购物车，添加记录，修改记录，删除记录，评论，登出等一系列我们熟悉的操作。这些操作都被记录在日志信息里面。我们要对日志信息进行分析。

本文中，我们对购买东西和加入购物车两个行为进行分析。然后生成相应的报表，这样我们可以通过报表查看用户在什么时候喜欢购买东西，什么时候喜欢加入购物车，从而，在相应的时间采取行动，激烈用户购买东西，推荐商品给用户加入购物车（加入购物车，这属于潜在购买用户）。

毕竟网站盈利才是我们希望达到的目的，对吧。

 

1.抽象用户行为

  1.     // 用户的action
  2.     public static final String[] USER_ACTION = { "Register", "Login", "View", "Click", "Double_Click", "Buy", "Shopping_Car", "Add", "Edit", "Delete", "Comment", "Logout" };

 

2.日志格式定义

  1. 115.19.62.102    海南    2018-12-20    1545286960749    1735787074662918890    www.hongten.com    Edit
  2. 27.177.45.84    新疆    2018-12-20    1545286962255    6667636903937987930    www.hongten.com    Delete
  3. 176.54.120.96    宁夏    2018-12-20    1545286962256    6988408478348165495    www.hongten.com    Comment
  4. 175.117.33.187    辽宁    2018-12-20    1545286962257    8411202446705338969    www.hongten.com    Shopping_Car
  5. 17.67.62.213    天津    2018-12-20    1545286962258    7787584752786413943    www.hongten.com    Add
  6. 137.81.41.9    海南    2018-12-20    1545286962259    6218367085234099455    www.hongten.com    Shopping_Car
  7. 125.187.107.57    山东    2018-12-20    1545286962260    3358658811146151155    www.hongten.com    Double_Click
  8. 104.167.205.87    内蒙    2018-12-20    1545286962261    2303468282544965471    www.hongten.com    Shopping_Car
  9. 64.106.149.83    河南    2018-12-20    1545286962262    8422202443986582525    www.hongten.com    Delete
  10. 138.22.156.183    浙江    2018-12-20    1545286962263    7649154147863130337    www.hongten.com    Shopping_Car
  11. 41.216.103.31    河北    2018-12-20    1545286962264    6785302169446728008    www.hongten.com    Shopping_Car
  12. 132.144.93.20    广东    2018-12-20    1545286962265    6444575166009004406    www.hongten.com    Add

日志格式：

  1. //log fromat
  2. String log = ip + "\t" + address + "\t" + d + "\t" + timestamp + "\t" + userid + "\t" + Common.WEB_SITE + "\t" + action;

 

3.系统架构

 

 

4.报表样式

由于我采用的是随机生成数据，所有，我们看到的结果呈现线性增长

这里我只是实现了一个小时的报表，当然，也可以做一天，一个季度，全年，三年，五年的报表，可以根据实际需求实现即可。

 

5.组件分布情况

我总共搭建了4个节点node1,node2,node3,node4（注： 4个节点上面都要有JDK）

Zookeeper安装在node1,node2,nod3

Hadoop集群在node1,node2,nod3,node4

Hbase集群在node1,node2,nod3,node4

Flume安装在node2

Kafka安装在node1,node2,node3

Storm安装在node1,node2,node3

 

6.具体实现

6.1.配置Flume

  1. --从node2
  2. cd flumedir
  4. vi flume2kafka
  6. --node2配置如下
  7. a1.sources = r1
  8. a1.sinks = k1
  9. a1.channels = c1
  11. # Describe/configure the source
  12. a1.sources.r1.type = avro
  13. a1.sources.r1.bind = node2
  14. a1.sources.r1.port = 41414
  16. # Describe the sink
  17. a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  18. a1.sinks.k1.topic = all_my_log
  19. a1.sinks.k1.brokerList = node1:9092,node2:9092,node3:9092
  20. a1.sinks.k1.requiredAcks = 1
  21. a1.sinks.k1.batchSize = 20
  23. # Use a channel which buffers events in memory
  24. a1.channels.c1.type = memory
  25. a1.channels.c1.capacity = 1000000
  26. a1.channels.c1.transactionCapacity = 10000
  28. # Bind the source and sink to the channel
  29. a1.sources.r1.channels = c1
  30. a1.sinks.k1.channel = c1
  32. :wq

 

6.2.启动Zookeeper

  1. --关闭防火墙node1,node2,node3,node4
  2. service iptables stop
  4. --启动Zookeeper,在node1，node2，node3
  5. zkServer.sh start

 

6.3.启动Kafka

  1. --启动kafka
  2. --分别进入node1,node2,node3
  3. cd /root/kafka/kafka_2.10-0.8.2.2
  4. ./start-kafka.sh

 

6.4.启动Flume服务

  1. --进入node2，启动
  2. cd /root/flumedir
  3. flume-ng agent -n a1 -c conf -f flume2kafka -Dflume.root.logger=DEBUG,console

 

6.5.产生日志信息并写入到Flume

运行java 代码，产生日志信息并写入到Flume服务器

  1. package com.b510.big.data.flume.client;
  3. import java.nio.charset.Charset;
  4. import java.text.SimpleDateFormat;
  5. import java.util.Date;
  6. import java.util.Random;
  7. import java.util.concurrent.ExecutorService;
  8. import java.util.concurrent.Executors;
  9. import java.util.concurrent.TimeUnit;
  11. import org.apache.flume.Event;
  12. import org.apache.flume.EventDeliveryException;
  13. import org.apache.flume.api.RpcClient;
  14. import org.apache.flume.api.RpcClientFactory;
  15. import org.apache.flume.event.EventBuilder;
  17. /**
  18.  * @author Hongten
  19.  * 
  20.  *         功能： 模拟产生用户日志信息，并且向Flume发送数据
  21.  */
  22. public class FlumeClient {
  24.     public static void main(String[] args) {
  25.         ExecutorService exec = Executors.newCachedThreadPool();
  26.         exec.execute(new GenerateDataAndSend2Flume());
  28.         exec.shutdown();
  29.     }
  31. }
  33. class GenerateDataAndSend2Flume implements Runnable {
  35.     FlumeRPCClient flumeRPCClient;
  36.     static Random random = new Random();
  38.     GenerateDataAndSend2Flume() {
  39.         // 初始化RPC客户端
  40.         flumeRPCClient = new FlumeRPCClient();
  41.         flumeRPCClient.init(Common.FLUME_HOST_NAME, Common.FLUME_PORT);
  42.     }
  44.     @Override
  45.     public void run() {
  46.         while (true) {
  47.             Date date = new Date();
  48.             SimpleDateFormat simpleDateFormat = new SimpleDateFormat(Common.DATE_FORMAT_YYYYDDMM);
  49.             String d = simpleDateFormat.format(date);
  50.             Long timestamp = new Date().getTime();
  51.             // ip地址生成
  52.             String ip = random.nextInt(Common.MAX_IP_NUMBER) + "." + random.nextInt(Common.MAX_IP_NUMBER) + "." + random.nextInt(Common.MAX_IP_NUMBER) + "." + random.nextInt(Common.MAX_IP_NUMBER);
  53.             // ip地址对应的address(这里是为了构造数据，并没有按照真实的ip地址，找到对应的address)
  54.             String address = Common.ADDRESS[random.nextInt(Common.ADDRESS.length)];
  56.             Long userid = Math.abs(random.nextLong());
  57.             String action = Common.USER_ACTION[random.nextInt(Common.USER_ACTION.length)];
  58.             // 日志信息构造
  59.             // example : 199.80.45.117 云南 2018-12-20 1545285957720 3086250439781555145 www.hongten.com Buy
  60.             String data = ip + "\t" + address + "\t" + d + "\t" + timestamp + "\t" + userid + "\t" + Common.WEB_SITE + "\t" + action;
  61.             //System.out.println(data);
  63.             // 往Flume发送数据
  64.             flumeRPCClient.sendData2Flume(data);
  66.             try {
  67.                 TimeUnit.MICROSECONDS.sleep(random.nextInt(1000));
  68.             } catch (InterruptedException e) {
  69.                 flumeRPCClient.cleanUp();
  70.                 System.out.println("interrupted exception : " + e);
  71.             }
  72.         }
  73.     }
  74. }
  76. class FlumeRPCClient {
  78.     private RpcClient client;
  79.     private String hostname;
  80.     private int port;
  82.     public void init(String hostname, int port) {
  83.         this.hostname = hostname;
  84.         this.port = port;
  85.         this.client = getRpcClient(hostname, port);
  86.     }
  88.     public void sendData2Flume(String data) {
  89.         Event event = EventBuilder.withBody(data, Charset.forName(Common.CHAR_FORMAT));
  91.         try {
  92.             client.append(event);
  93.         } catch (EventDeliveryException e) {
  94.             cleanUp();
  95.             client = null;
  96.             client = getRpcClient(hostname, port);
  97.         }
  98.     }
  100.     public RpcClient getRpcClient(String hostname, int port) {
  101.         return RpcClientFactory.getDefaultInstance(hostname, port);
  102.     }
  104.     public void cleanUp() {
  105.         // Close the RPC connection
  106.         client.close();
  107.     }
  108. }
  110. // 所有的常量定义
  111. class Common {
  112.     public static final String CHAR_FORMAT = "UTF-8";
  114.     public static final String DATE_FORMAT_YYYYDDMM = "yyyy-MM-dd";
  116.     // this is a test web site
  117.     public static final String WEB_SITE = "www.hongten.com";
  119.     // 用户的action
  120.     public static final String[] USER_ACTION = { "Register", "Login", "View", "Click", "Double_Click", "Buy", "Shopping_Car", "Add", "Edit", "Delete", "Comment", "Logout" };
  122.     public static final int MAX_IP_NUMBER = 224;
  123.     // ip所对应的地址
  124.     public static String[] ADDRESS = { "北京", "天津", "上海", "广东", "重庆", "河北", "山东", "河南", "云南", "山西", "甘肃", "安徽", "福建", "黑龙江", "海南", "四川", "贵州", "宁夏", "新疆", "湖北", "湖南", "山西", "辽宁", "吉林", "江苏", "浙江", "青海", "江西", "西藏", "内蒙", "广西", "香港", "澳门", "台湾", };
  126.     // Flume conf
  127.     public static final String FLUME_HOST_NAME = "node2";
  128.     public static final int FLUME_PORT = 41414;
  130. }

 

6.6.监听Kafka

  1. --进入node3，启动kafka消费者
  2. cd /home/kafka-2.10/bin
  3. ./kafka-console-consumer.sh --zookeeper node1,node2,node3 --from-beginning --topic all_my_log

 

运行效果：

  1. 168.208.193.207    安徽    2018-12-20    1545287646527    5462770148222682599    www.hongten.com    Login
  2. 103.143.79.127    新疆    2018-12-20    1545287646529    3389475301916412717    www.hongten.com    Login
  3. 111.208.80.39    山东    2018-12-20    1545287646531    535601622597096753    www.hongten.com    Shopping_Car
  4. 105.30.86.46    四川    2018-12-20    1545287646532    7825340079790811845    www.hongten.com    Login
  5. 205.55.33.74    新疆    2018-12-20    1545287646533    4228838365367235561    www.hongten.com    Logout
  6. 34.44.60.134    安徽    2018-12-20    1545287646536    702584874247456732    www.hongten.com    Double_Click
  7. 154.169.15.145    广东    2018-12-20    1545287646537    1683351753576425036    www.hongten.com    View
  8. 126.28.192.28    湖南    2018-12-20    1545287646538    8319814684518483148    www.hongten.com    Edit
  9. 5.140.156.73    台湾    2018-12-20    1545287646539    7432409906375230025    www.hongten.com    Logout
  10. 72.175.210.95    西藏    2018-12-20    1545287646540    5233707593244910849    www.hongten.com    View
  11. 121.25.190.25    广西    2018-12-20    1545287646541    268200251881841673    www.hongten.com    Buy

 

6.7.在Kafka创建Topic

  1. --进入node1，创建一个topic：filtered_log
  2. --设置3个partitions
  3. --replication-factor=3
  4. ./kafka-topics.sh --zookeeper node1,node2,node3 --create --topic filtered_log --partitions 3 --replication-factor 3

 

6.8.Storm清洗数据

  • Storm从Kafka消费数据
  • Storm对数据进行筛选(Buy-已经购买,Shopping_Car-潜在购买)
  • Storm把筛选的数据放入到Kafka
  1. package com.b510.big.data.storm.process;
  3. import java.util.ArrayList;
  4. import java.util.List;
  5. import java.util.Properties;
  7. import storm.kafka.KafkaSpout;
  8. import storm.kafka.SpoutConfig;
  9. import storm.kafka.StringScheme;
  10. import storm.kafka.ZkHosts;
  11. import storm.kafka.bolt.KafkaBolt;
  12. import storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
  13. import storm.kafka.bolt.selector.DefaultTopicSelector;
  14. import backtype.storm.Config;
  15. import backtype.storm.LocalCluster;
  16. import backtype.storm.StormSubmitter;
  17. import backtype.storm.generated.AlreadyAliveException;
  18. import backtype.storm.generated.InvalidTopologyException;
  19. import backtype.storm.spout.SchemeAsMultiScheme;
  20. import backtype.storm.topology.BasicOutputCollector;
  21. import backtype.storm.topology.OutputFieldsDeclarer;
  22. import backtype.storm.topology.TopologyBuilder;
  23. import backtype.storm.topology.base.BaseBasicBolt;
  24. import backtype.storm.tuple.Fields;
  25. import backtype.storm.tuple.Tuple;
  26. import backtype.storm.tuple.Values;
  28. public class LogFilterTopology {
  30.     public static void main(String[] args) {
  32.         ZkHosts zkHosts = new ZkHosts(Common.ZOOKEEPER_QUORUM);
  33.         //Spout从'filtered_log' topic里面获取数据
  34.         SpoutConfig spoutConfig = new SpoutConfig(zkHosts, Common.ALL_MY_LOG_TOPIC, Common.ZOOKEEPER_ROOT, Common.ZOOKEEPER_ID);
  35.         List<String> zkServers = new ArrayList<>();
  36.         for (String host : zkHosts.brokerZkStr.split(",")) {
  37.             zkServers.add(host.split(":")[0]);
  38.         }
  40.         spoutConfig.zkServers = zkServers;
  41.         spoutConfig.zkPort = Common.ZOOKEEPER_PORT;
  42.         spoutConfig.forceFromStart = true;
  43.         spoutConfig.socketTimeoutMs = 60 * 60 * 1000;
  44.         spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
  46.         // 创建KafkaSpout
  47.         KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
  49.         TopologyBuilder builder = new TopologyBuilder();
  50.         // Storm从Kafka消费数据
  51.         builder.setSpout(Common.KAFKA_SPOUT, kafkaSpout, 3);
  52.         // Storm对数据进行筛选(Buy-已经购买,Shopping_Car-潜在购买)
  53.         builder.setBolt(Common.FILTER_BOLT, new FilterBolt(), 8).shuffleGrouping(Common.KAFKA_SPOUT);
  55.         // 创建KafkaBolt
  56.         @SuppressWarnings({ "unchecked", "rawtypes" })
  57.         KafkaBolt kafkaBolt = new KafkaBolt().withTopicSelector(new DefaultTopicSelector(Common.FILTERED_LOG_TOPIC)).withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper());
  59.         // Storm把筛选的数据放入到Kafka
  60.         builder.setBolt(Common.KAFKA_BOLT, kafkaBolt, 2).shuffleGrouping(Common.FILTER_BOLT);
  62.         Properties props = new Properties();
  63.         props.put("metadata.broker.list", Common.STORM_METADATA_BROKER_LIST);
  64.         props.put("request.required.acks", Common.STORM_REQUEST_REQUIRED_ACKS);
  65.         props.put("serializer.class", Common.STORM_SERILIZER_CLASS);
  67.         Config conf = new Config();
  68.         conf.put("kafka.broker.properties", props);
  70.         conf.put(Config.STORM_ZOOKEEPER_SERVERS, zkServers);
  72.         if (args == null || args.length == 0) {
  73.             // 本地方式运行
  74.             LocalCluster localCluster = new LocalCluster();
  75.             localCluster.submitTopology("storm-kafka-topology", conf, builder.createTopology());
  76.         } else {
  77.             // 集群方式运行
  78.             conf.setNumWorkers(3);
  79.             try {
  80.                 StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
  81.             } catch (AlreadyAliveException | InvalidTopologyException e) {
  82.                 System.out.println("error : " + e);
  83.             }
  84.         }
  85.     }
  86. }
  88. class FilterBolt extends BaseBasicBolt {
  90.     private static final long serialVersionUID = 1L;
  92.     @Override
  93.     public void execute(Tuple input, BasicOutputCollector collector) {
  94.         String logStr = input.getString(0);
  95.         // 只针对我们感兴趣的关键字进行过滤
  96.         // 这里我们过滤包含'Buy', 'Shopping_Car'的日志信息
  97.         if (logStr.contains(Common.KEY_WORD_BUY) || logStr.contains(Common.KEY_WORD_SHOPPING_CAR)) {
  98.             collector.emit(new Values(logStr));
  99.         }
  100.     }
  102.     @Override
  103.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  104.         declarer.declare(new Fields(FieldNameBasedTupleToKafkaMapper.BOLT_MESSAGE));
  105.     }
  106. }
  108. class Common {
  109.     public static final String ALL_MY_LOG_TOPIC = "all_my_log";
  110.     public static final String FILTERED_LOG_TOPIC = "filtered_log";
  111.     
  112.     public static final String DATE_FORMAT_YYYYDDMMHHMMSS = "yyyyMMddHHmmss";
  113.     public static final String DATE_FORMAT_HHMMSS = "HHmmss";
  114.     public static final String DATE_FORMAT_HHMMSS_DEFAULT_VALUE = "000001";
  116.     public static final String HBASE_ZOOKEEPER_LIST = "node1:2888,node2:2888,node3:2888";
  117.     public static final int ZOOKEEPER_PORT = 2181;
  118.     public static final String ZOOKEEPER_QUORUM = "node1:" + ZOOKEEPER_PORT + ",node2:" + ZOOKEEPER_PORT + ",node3:" + ZOOKEEPER_PORT + "";
  119.     public static final String ZOOKEEPER_ROOT = "/MyKafka";
  120.     public static final String ZOOKEEPER_ID = "MyTrack";
  122.     public static final String KAFKA_SPOUT = "kafkaSpout";
  123.     public static final String FILTER_BOLT = "filterBolt";
  124.     public static final String PROCESS_BOLT = "processBolt";
  125.     public static final String HBASE_BOLT = "hbaseBolt";
  126.     public static final String KAFKA_BOLT = "kafkaBolt";
  128.     // Storm Conf
  129.     public static final String STORM_METADATA_BROKER_LIST = "node1:9092,node2:9092,node3:9092";
  130.     public static final String STORM_REQUEST_REQUIRED_ACKS = "1";
  131.     public static final String STORM_SERILIZER_CLASS = "kafka.serializer.StringEncoder";
  133.     // key word
  134.     public static final String KEY_WORD_BUY = "Buy";
  135.     public static final String KEY_WORD_SHOPPING_CAR = "Shopping_Car";
  136.     
  137.     //hbase
  138.     public static final String TABLE_USER_ACTION = "t_user_actions";
  139.     public static final String COLUMN_FAMILY = "cf";
  140.     //间隔多少秒写入Hbase一次
  141.     public static final int WRITE_RECORD_TO_TABLE_PER_SECOND = 1;
  142.     public static final int TABLE_MAX_VERSION = (60/WRITE_RECORD_TO_TABLE_PER_SECOND) * 60 * 24;
  143. }

 

6.9.监听Kafka

  1. --进入node3，启动kafka消费者
  2. cd /home/kafka-2.10/bin
  3. ./kafka-console-consumer.sh --zookeeper node1,node2,node3 --from-beginning --topic filtered_log

 

效果：

  1. 87.26.135.185    黑龙江    2018-12-20    1545290594658    7290881731606227972    www.hongten.com    Shopping_Car
  2. 60.96.96.38    青海    2018-12-20    1545290594687    6935901257286057015    www.hongten.com    Shopping_Car
  3. 43.159.110.193    江苏    2018-12-20    1545290594727    7096698224110515553    www.hongten.com    Shopping_Car
  4. 21.103.139.11    山西    2018-12-20    1545290594693    7805867078876194442    www.hongten.com    Shopping_Car
  5. 139.51.213.184    广东    2018-12-20    1545290594729    8048796865619113514    www.hongten.com    Buy
  6. 58.213.148.89    河北    2018-12-20    1545290594708    5176551342435592748    www.hongten.com    Buy
  7. 36.205.221.116    湖南    2018-12-20    1545290594715    4484717918039766421    www.hongten.com    Shopping_Car
  8. 135.194.103.53    北京    2018-12-20    1545290594769    4833011508087432349    www.hongten.com    Shopping_Car
  9. 180.21.100.66    贵州    2018-12-20    1545290594752    5270357330431599426    www.hongten.com    Buy
  10. 167.71.65.70    山西    2018-12-20    1545290594790    275898530145861990    www.hongten.com    Buy
  11. 125.51.21.199    宁夏    2018-12-20    1545290594814    3613499600574777198    www.hongten.com    Buy

 

 

6.10.Storm再次消费Kafka数据处理后保存数据到Hbase

  • Storm再次从Kafka消费数据
  • Storm对数据进行统计（Buy-已经购买人数,Shopping_Car-潜在购买人数）
  • Storm将数据写入到Hbase
  1. package com.b510.big.data.storm.process;
  3. import java.io.IOException;
  4. import java.text.SimpleDateFormat;
  5. import java.util.ArrayList;
  6. import java.util.Date;
  7. import java.util.List;
  8. import java.util.Map;
  9. import java.util.Properties;
  11. import org.apache.hadoop.conf.Configuration;
  12. import org.apache.hadoop.hbase.HColumnDescriptor;
  13. import org.apache.hadoop.hbase.HTableDescriptor;
  14. import org.apache.hadoop.hbase.TableName;
  15. import org.apache.hadoop.hbase.client.HBaseAdmin;
  16. import org.apache.hadoop.hbase.client.HConnection;
  17. import org.apache.hadoop.hbase.client.HConnectionManager;
  18. import org.apache.hadoop.hbase.client.HTableInterface;
  19. import org.apache.hadoop.hbase.client.Put;
  21. import storm.kafka.KafkaSpout;
  22. import storm.kafka.SpoutConfig;
  23. import storm.kafka.StringScheme;
  24. import storm.kafka.ZkHosts;
  25. import backtype.storm.Config;
  26. import backtype.storm.LocalCluster;
  27. import backtype.storm.StormSubmitter;
  28. import backtype.storm.generated.AlreadyAliveException;
  29. import backtype.storm.generated.InvalidTopologyException;
  30. import backtype.storm.spout.SchemeAsMultiScheme;
  31. import backtype.storm.task.TopologyContext;
  32. import backtype.storm.topology.BasicOutputCollector;
  33. import backtype.storm.topology.IBasicBolt;
  34. import backtype.storm.topology.OutputFieldsDeclarer;
  35. import backtype.storm.topology.TopologyBuilder;
  36. import backtype.storm.topology.base.BaseBasicBolt;
  37. import backtype.storm.tuple.Fields;
  38. import backtype.storm.tuple.Tuple;
  39. import backtype.storm.tuple.Values;
  41. public class LogProcessTopology {
  43.     public static void main(String[] args) {
  45.         ZkHosts zkHosts = new ZkHosts(Common.ZOOKEEPER_QUORUM);
  46.         //Spout从'filtered_log' topic里面获取数据
  47.         SpoutConfig spoutConfig = new SpoutConfig(zkHosts, Common.FILTERED_LOG_TOPIC, Common.ZOOKEEPER_ROOT, Common.ZOOKEEPER_ID);
  48.         List<String> zkServers = new ArrayList<>();
  49.         for (String host : zkHosts.brokerZkStr.split(",")) {
  50.             zkServers.add(host.split(":")[0]);
  51.         }
  53.         spoutConfig.zkServers = zkServers;
  54.         spoutConfig.zkPort = Common.ZOOKEEPER_PORT;
  55.         spoutConfig.forceFromStart = true;
  56.         spoutConfig.socketTimeoutMs = 60 * 60 * 1000;
  57.         spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
  59.         // 创建KafkaSpout
  60.         KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
  62.         TopologyBuilder builder = new TopologyBuilder();
  63.         // Storm再次从Kafka消费数据
  64.         builder.setSpout(Common.KAFKA_SPOUT, kafkaSpout, 3);
  65.         // Storm对数据进行统计（Buy-已经购买人数,Shopping_Car-潜在购买人数）
  66.         builder.setBolt(Common.PROCESS_BOLT, new ProcessBolt(), 3).shuffleGrouping(Common.KAFKA_SPOUT);
  67.         // Storm将数据写入到Hbase
  68.         builder.setBolt(Common.HBASE_BOLT, new HbaseBolt(), 3).shuffleGrouping(Common.PROCESS_BOLT);
  70.         Properties props = new Properties();
  71.         props.put("metadata.broker.list", Common.STORM_METADATA_BROKER_LIST);
  72.         props.put("request.required.acks", Common.STORM_REQUEST_REQUIRED_ACKS);
  73.         props.put("serializer.class", Common.STORM_SERILIZER_CLASS);
  75.         Config conf = new Config();
  76.         conf.put("kafka.broker.properties", props);
  78.         conf.put(Config.STORM_ZOOKEEPER_SERVERS, zkServers);
  80.         if (args == null || args.length == 0) {
  81.             // 本地方式运行
  82.             LocalCluster localCluster = new LocalCluster();
  83.             localCluster.submitTopology("storm-kafka-topology", conf, builder.createTopology());
  84.         } else {
  85.             // 集群方式运行
  86.             conf.setNumWorkers(3);
  87.             try {
  88.                 StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
  89.             } catch (AlreadyAliveException | InvalidTopologyException e) {
  90.                 System.out.println("error : " + e);
  91.             }
  92.         }
  93.         
  94.     }
  95. }
  97. class ProcessBolt extends BaseBasicBolt {
  99.     private static final long serialVersionUID = 1L;
  101.     @Override
  102.     public void execute(Tuple input, BasicOutputCollector collector) {
  103.         String logStr = input.getString(0);
  104.         if (logStr != null) {
  105.             String infos[] = logStr.split("\\t");
  106.             //180.21.100.66    贵州    2018-12-20    1545290594752    5270357330431599426    www.hongten.com    Buy
  107.             collector.emit(new Values(infos[2], infos[6]));
  108.         }
  109.     }
  111.     @Override
  112.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  113.         declarer.declare(new Fields("date", "user_action"));
  114.     }
  115. }
  117. class HbaseBolt implements IBasicBolt {
  118.     private static final long serialVersionUID = 1L;
  120.     HBaseDAO hBaseDAO = null;
  121.     
  122.     SimpleDateFormat simpleDateFormat = null;
  123.     SimpleDateFormat simpleDateFormatHHMMSS = null;
  124.     
  125.     int userBuyCount = 0;
  126.     int userShoopingCarCount = 0;
  127.     
  128.     //这里要考虑避免频繁写入数据到hbase
  129.     int writeToHbaseMaxNum = Common.WRITE_RECORD_TO_TABLE_PER_SECOND * 1000;
  130.     long begin = System.currentTimeMillis();
  131.     long end = 0;
  132.     
  133.     @SuppressWarnings("rawtypes")
  134.     @Override
  135.     public void prepare(Map map, TopologyContext context) {
  136.         hBaseDAO = new HBaseDAOImpl();
  137.         simpleDateFormat = new SimpleDateFormat(Common.DATE_FORMAT_YYYYDDMMHHMMSS);
  138.         simpleDateFormatHHMMSS = new SimpleDateFormat(Common.DATE_FORMAT_HHMMSS);
  139.         hBaseDAO.createTable(Common.TABLE_USER_ACTION, new String[]{Common.COLUMN_FAMILY}, Common.TABLE_MAX_VERSION);
  140.     }
  142.     @Override
  143.     public void execute(Tuple input, BasicOutputCollector collector) {
  144.         // 如果时间是第二天的凌晨1s
  145.         // 需要对count做清零处理
  146.         //不过这里的判断不是很准确，因为在此时，可能前一天的数据还没有处理完
  147.         if (simpleDateFormatHHMMSS.format(new Date()).equals(Common.DATE_FORMAT_HHMMSS_DEFAULT_VALUE)) {
  148.             userBuyCount = 0;
  149.             userShoopingCarCount = 0;
  150.         }
  151.         
  152.         if (input != null) {
  153.             // base one ProcessBolt.declareOutputFields()
  154.             String date = input.getString(0);
  155.             String userAction = input.getString(1);
  157.             if (userAction.equals(Common.KEY_WORD_BUY)) {
  158.                 //同一个user在一天之内可以重复'Buy'动作
  159.                 userBuyCount++;
  160.             }
  162.             if (userAction.equals(Common.KEY_WORD_SHOPPING_CAR)) {
  163.                 userShoopingCarCount++;
  164.             }
  166.             end = System.currentTimeMillis();
  167.             if ((end - begin) > writeToHbaseMaxNum) {
  168.                 System.out.println("hbase_key: " + Common.KEY_WORD_BUY + "_" + date + " , userBuyCount: " + userBuyCount + ", userShoopingCarCount :" + userShoopingCarCount);
  169.                 
  170.                 //往hbase中写入数据
  171.                 String quailifer = simpleDateFormat.format(new Date());
  172.                 hBaseDAO.insert(Common.TABLE_USER_ACTION , 
  173.                         Common.KEY_WORD_BUY + "_" + date, 
  174.                         Common.COLUMN_FAMILY, 
  175.                         new String[] { quailifer },
  176.                         new String[] { "{user_buy_count:" + userBuyCount + "}" }
  177.                         );
  178.                 hBaseDAO.insert(Common.TABLE_USER_ACTION , 
  179.                         Common.KEY_WORD_SHOPPING_CAR + "_" + date, 
  180.                         Common.COLUMN_FAMILY, 
  181.                         new String[] { quailifer },
  182.                         new String[] { "{user_shopping_car_count:" + userShoopingCarCount + "}" }
  183.                         );
  184.                 begin = System.currentTimeMillis();
  185.             }
  186.         }
  187.     }
  189.     @Override
  190.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  192.     }
  194.     @Override
  195.     public Map<String, Object> getComponentConfiguration() {
  196.         return null;
  197.     }
  199.     @Override
  200.     public void cleanup() {
  202.     }
  203. }
  205. interface HBaseDAO {
  206.     public void createTable(String tableName, String[] columnFamilys, int maxVersion);
  207.     public void insert(String tableName, String rowKey, String family, String quailifer[], String value[]);
  208. }
  210. class HBaseDAOImpl implements HBaseDAO {
  212.     HConnection hConnection = null;
  213.     static Configuration conf = null;
  215.     public HBaseDAOImpl() {
  216.         conf = new Configuration();
  217.         conf.set("hbase.zookeeper.quorum", Common.HBASE_ZOOKEEPER_LIST);
  218.         try {
  219.             hConnection = HConnectionManager.createConnection(conf);
  220.         } catch (IOException e) {
  221.             e.printStackTrace();
  222.         }
  223.     }
  224.     
  225.     public void createTable(String tableName, String[] columnFamilys, int maxVersion) {
  226.         try {
  227.             HBaseAdmin admin = new HBaseAdmin(conf);
  228.             if (admin.tableExists(tableName)) {
  229.                 System.err.println("table existing in hbase.");
  230.             } else {
  231.                 HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
  232.                 for (String columnFamily : columnFamilys) {
  233.                     HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(columnFamily);
  234.                     hColumnDescriptor.setMaxVersions(maxVersion);
  235.                     tableDesc.addFamily(hColumnDescriptor);
  236.                 }
  238.                 admin.createTable(tableDesc);
  239.                 System.err.println("table is created.");
  240.             }
  241.             admin.close();
  242.         } catch (Exception e) {
  243.             e.printStackTrace();
  244.         }
  245.     }
  246.     
  247.     @Override
  248.     public void insert(String tableName, String rowKey, String family, String quailifer[], String value[]) {
  249.         HTableInterface table = null;
  250.         try {
  251.             table = hConnection.getTable(tableName);
  252.             Put put = new Put(rowKey.getBytes());
  253.             for (int i = 0; i < quailifer.length; i++) {
  254.                 String col = quailifer[i];
  255.                 String val = value[i];
  256.                 put.add(family.getBytes(), col.getBytes(), val.getBytes());
  257.             }
  258.             table.put(put);
  259.             System.err.println("save record successfuly.");
  260.         } catch (Exception e) {
  261.             e.printStackTrace();
  262.         } finally {
  263.             try {
  264.                 table.close();
  265.             } catch (IOException e) {
  266.                 e.printStackTrace();
  267.             }
  268.         }
  269.     }
  271. }

 

Storm处理逻辑：

1.每秒向Hbase写入数据

2.明天凌晨会重置数据

如果，我们一直运行上面的程序，那么，系统就会一直往Hbase里面写入数据，那么这样，我们就可以采集到我们生成报表的数据了。

那么下面就是报表实现

 

6.11.读取Hbase数据通过POI生成Excel Report

  • 读取Hbase数据
  • 通过POI生成Excel报表
  1. package com.b510.big.data.poi;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.FileOutputStream;
  6. import java.io.IOException;
  7. import java.io.InputStream;
  8. import java.util.ArrayList;
  9. import java.util.List;
  11. import org.apache.hadoop.conf.Configuration;
  12. import org.apache.hadoop.hbase.Cell;
  13. import org.apache.hadoop.hbase.CellUtil;
  14. import org.apache.hadoop.hbase.client.Get;
  15. import org.apache.hadoop.hbase.client.HConnection;
  16. import org.apache.hadoop.hbase.client.HConnectionManager;
  17. import org.apache.hadoop.hbase.client.HTableInterface;
  18. import org.apache.hadoop.hbase.client.Result;
  19. import org.apache.poi.xssf.usermodel.XSSFCell;
  20. import org.apache.poi.xssf.usermodel.XSSFSheet;
  21. import org.apache.poi.xssf.usermodel.XSSFWorkbook;
  23. public class ReportUtil {
  25.     public static void main(String[] args) throws Exception {
  27.         String year = "2018";
  28.         String month = "12";
  29.         String day = "21";
  30.         String hour = "14";
  32.         generateReport(year, month, day, hour);
  33.     }
  35.     private static void generateReport(String year, String month, String day, String hour) {
  36.         HBaseDAO hBaseDAO = new HBaseDAOImpl();
  37.         // format: yyyyMMddHH
  38.         String begin = year + month + day + hour;
  39.         String[] split = generateQuailifers(begin);
  41.         List<Integer> userBuyCountList = getData(hBaseDAO, year, month, day, split, Common.KEY_WORD_BUY);
  42.         List<Integer> userShoppingCarCountList = getData(hBaseDAO, year, month, day, split, Common.KEY_WORD_SHOPPING_CAR);
  44.         //System.err.println(userBuyCountList.size());
  45.         //System.err.println(userShoppingCarCountList.size());
  47.         writeExcel(year, month, day, hour, userBuyCountList, userShoppingCarCountList);
  48.     }
  50.     private static void writeExcel(String year, String month, String day, String hour, List<Integer> userBuyCountList, List<Integer> userShoppingCarCountList) {
  51.         try {
  52.             File file = new File(Common.REPORT_TEMPLATE);
  53.             InputStream in = new FileInputStream(file);
  54.             XSSFWorkbook wb = new XSSFWorkbook(in);
  55.             XSSFSheet sheet = wb.getSheetAt(0);
  56.             if (sheet != null) {
  57.                 XSSFCell cell = null;
  59.                 cell = sheet.getRow(0).getCell(0);
  60.                 cell.setCellValue("One Hour Report-" + year + "-" + month + "-" + day + " From " + hour + ":00 To " + hour + ":59");
  62.                 putData(userBuyCountList, sheet, 3);
  63.                 putData(userShoppingCarCountList, sheet, 7);
  65.                 FileOutputStream out = new FileOutputStream(Common.REPORT_ONE_HOUR);
  66.                 wb.write(out);
  67.                 out.close();
  68.                 System.err.println("done.");
  69.             }
  70.         } catch (Exception e) {
  71.             System.err.println("Exception" + e);
  72.         }
  73.     }
  75.     private static void putData(List<Integer> userBuyCountList, XSSFSheet sheet, int rowNum) {
  76.         XSSFCell cell;
  77.         if (userBuyCountList != null && userBuyCountList.size() > 0) {
  78.             for (int i = 0; i < userBuyCountList.size(); i++) {
  79.                 cell = sheet.getRow(rowNum).getCell(i + 1);
  80.                 cell.setCellValue(userBuyCountList.get(i));
  81.             }
  82.         }
  83.     }
  85.     private static List<Integer> getData(HBaseDAO hBaseDAO, String year, String month, String day, String[] split, String preKey) {
  86.         List<Integer> list = new ArrayList<Integer>();
  87.         Result rs = hBaseDAO.getOneRowAndMultiColumn(Common.TABLE_USER_ACTION, preKey + "_" + year + "-" + month + "-" + day, split);
  88.         for (Cell cell : rs.rawCells()) {
  89.             String value = new String(CellUtil.cloneValue(cell)).split(":")[1].trim();
  90.             value = value.substring(0, value.length() - 1);
  91.             list.add(Integer.valueOf(value));
  92.         }
  93.         return list;
  94.     }
  96.     private static String[] generateQuailifers(String begin) {
  97.         StringBuilder sb = new StringBuilder();
  98.         for (int i = 0; i < 60;) {
  100.             if (i == 0 || i == 5) {
  101.                 sb.append(begin).append("0").append(i).append("00").append(",");
  102.             } else {
  103.                 sb.append(begin).append(i).append("00").append(",");
  104.             }
  105.             i = i + 5;
  106.         }
  107.         sb.append(begin).append("5959");
  108.         String sbStr = sb.toString();
  109.         String[] split = sbStr.split(",");
  110.         return split;
  111.     }
  112. }
  114. interface HBaseDAO {
  115.     Result getOneRowAndMultiColumn(String tableName, String rowKey, String[] cols);
  116. }
  118. class HBaseDAOImpl implements HBaseDAO {
  120.     HConnection hConnection = null;
  121.     static Configuration conf = null;
  123.     public HBaseDAOImpl() {
  124.         conf = new Configuration();
  125.         conf.set("hbase.zookeeper.quorum", Common.HBASE_ZOOKEEPER_LIST);
  126.         try {
  127.             hConnection = HConnectionManager.createConnection(conf);
  128.         } catch (IOException e) {
  129.             e.printStackTrace();
  130.         }
  131.     }
  133.     @Override
  134.     public Result getOneRowAndMultiColumn(String tableName, String rowKey, String[] cols) {
  135.         HTableInterface table = null;
  136.         Result rsResult = null;
  137.         try {
  138.             table = hConnection.getTable(tableName);
  139.             Get get = new Get(rowKey.getBytes());
  140.             for (int i = 0; i < cols.length; i++) {
  141.                 get.addColumn(Common.COLUMN_FAMILY.getBytes(), cols[i].getBytes());
  142.             }
  143.             rsResult = table.get(get);
  144.         } catch (Exception e) {
  145.             e.printStackTrace();
  146.         } finally {
  147.             try {
  148.                 table.close();
  149.             } catch (IOException e) {
  150.                 e.printStackTrace();
  151.             }
  152.         }
  153.         return rsResult;
  154.     }
  156. }
  158. class Common {
  160.     // report
  161.     public static final String REPORT_TEMPLATE = "./resources/report.xlsx";
  162.     public static final String REPORT_ONE_HOUR = "./resources/one_report.xlsx";
  164.     public static final String DATE_FORMAT_YYYYDDMMHHMMSS = "yyyyMMddHHmmss";
  166.     public static final String HBASE_ZOOKEEPER_LIST = "node1:2888,node2:2888,node3:2888";
  168.     // key word
  169.     public static final String KEY_WORD_BUY = "Buy";
  170.     public static final String KEY_WORD_SHOPPING_CAR = "Shopping_Car";
  172.     // hbase
  173.     public static final String TABLE_USER_ACTION = "t_user_actions";
  174.     public static final String COLUMN_FAMILY = "cf";
  176. }

 

7.源码下载

Source Code：Flume_Kafka_Storm_Hbase_Hdfs_Poi_src.zip

相应的Jar文件，由于so big，自己根据import *信息加入。

 

8.总结

学习Big Data一段时间了，通过自己的学习和摸索，实现自己想要的应用，还是很有成就感的哈....当然，踩地雷也是一种不错的体验...:)

 

========================================================

More reading,and english is important.

I'm Hongten

 

  1. 大哥哥大姐姐，觉得有用打赏点哦！你的支持是我最大的动力。谢谢。
    Hongten博客排名在100名以内。粉丝过千。
    Hongten出品，必是精品。

E | hongtenzone@foxmail.com  B | http://www.cnblogs.com/hongten

========================================================

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Apache Kafka热门文章

  • Kafka监控工具汇总
  • Kraft模式下Kafka脚本的使用
  • Kafka web console安装
  • kafka_2.11-2.0.0_安装部署
  • ELK日志方案--使用Filebeat收集日志并输出到Kafka
  • Kafka相关内容总结（存储和性能）
  • 一次flume exec source采集日志到kafka因为单条日志数据非常大同步失败的踩坑带来的思考
  • Kafka使用jmxtrans+influxdb+grafana监控JMX指标
  • Micronaut 微服务中使用 Kafka
  • 什么是Kafka？

    • Apache Kafka推荐文章

  • 异源数据同步 → DataX 为什么要支持 kafka？
  • 解锁强强组合: 使用 Kafka + ClickHouse 快速搭建流数据实时处理平台(DoubleCloud 博客)
  • Kafka 线上性能调优
  • Kafka原理剖析之「位点提交」
  • Kafka 社区KIP-382中文译文（MirrorMaker2/集群复制/高可用/灾难恢复）
  • kafka源码阅读之MacBook Pro M1搭建Kafka2.7版本源码运行环境
  • Kafka干货之「零拷贝」
  • 扫盲Kafka？看这一篇就够了！
  • 从一次Kafka宕机说起（JVM hang）
  • 从Kafka中学习高性能系统如何设计

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号