devbox
花生_TL007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

电话日志分析callLog（一）_移动电话日志分析

作者：花生_TL007 | 2024-03-24 08:22:44

赞

踩

移动电话日志分析
  1. 一、项目简介
  2. ----------------------------------------------
  3.     1.hadoop+hbase+flume+zookeeper实现电信级海量通话日志数据的存储，随机
  4.     访问与实时读写。通过hash技术对rowkey进行分析处理，解决hbase的热点问题，协同
  5.     coprocessor，解决系统的高吞吐量和查询负载问题以及如何避免中间结果导致通知风暴或
  6.     死递归问题，让同学们体验到大数据技术在企业中实战应用
  7.  
  8.     2.整体架构分析
  9.       hadoop体系架构与ha配置方案。
  10.       hbase体系架构与ha配置方案
  11.       flume实时收集架构方案。
  12.       SSM实现前端web实现以及与后端HBase的交互架构方案。
  13.       hive+oozie实现的周期任务调度。
  14.       Spark streaming实现窗口化敏感词实时监控方案。
  15.  
  16.     3.hbase中callLogs表的设计与实现。
  17.       通话信息的内容分析与常用场景分析以及对rowkey的设计与实现。
  18.       重点讲解盐析的原理与热点问题的解决。rowkey的设计原则与实战
  19.       中的技巧。
  20.  
  21.     4.协处理原理与应用实战讲解。
  22.       被叫通话记录的设计思想讲解，以及通过协处理器方式实现callog日志主叫记录
  23.       被主换位与同步写入。在callog是表中数据的存储序列与双向查询方方式的一致
  24.       性透明结果处理。
  25.  
  26.     5.Hadoop以及HBase的HA集群配置与实战。
  27.       hadoop的使用QJM的高可用架构配置讲解，ResourceManager的高可用架构配置讲解。
  28.       zookeeper的工作原理以及配置、实操演练，hbase与Hadoop HA集成注意事项以及客户端
  29.       API编程细节处理。
  30.  
  31.  
  32. 二、创建新工程
  33. ------------------------------------------------
  34.     1.创建新工程 -- CallLogSystem
  35.  
  36.  
  37. 三、创建模拟日志生成程序模块CallLogGenModel
  38. -------------------------------------------------
  39.     1.创建模块 -- CallLogGenModel，添加Maven支持
  40.  
  41.     2.创建类calllog.gen.main.App.class
  42.  
  43.     3.编写App类
  44.     ---------------------------------------
  45.         
  1. package calllog.gen.main;
  2.  
  3.         import java.io.FileWriter;
  4.         import java.io.IOException;
  5.         import java.text.DecimalFormat;
  6.         import java.text.SimpleDateFormat;
  7.         import java.util.*;
  8.  
  9.         public class App {
  10.  
  11.             //电话簿
  12.             public static Map<String, String> callers = new HashMap<String, String>();
  13.             //电话号码
  14.             public static List<String> phoneNumbers = new ArrayList<String>();
  15.  
  16.             static{
  17.                 callers.put("15811111111", "史让");
  18.                 callers.put("18022222222", "赵嗄");
  19.                 callers.put("15133333333", "张锕 ");
  20.                 callers.put("13269364444", "王以");
  21.                 callers.put("15032295555", "张噢");
  22.                 callers.put("17731086666", "张类");
  23.                 callers.put("15338597777", "李平");
  24.                 callers.put("15733218888", "杜跑");
  25.                 callers.put("15614209999", "任阳");
  26.                 callers.put("15778421111", "梁鹏");
  27.                 callers.put("18641241111", "郭彤");
  28.                 callers.put("15732641111", "刘飞");
  29.                 callers.put("13341101111", "段星");
  30.                 callers.put("13560191111", "唐华");
  31.                 callers.put("18301581111", "杨谋");
  32.                 callers.put("13520401111", "温英");
  33.                 callers.put("18332561111", "朱宽");
  34.                 callers.put("18620191111", "刘宗");
  35.                 phoneNumbers.addAll(callers.keySet());
  36.  
  37.             }
  38.  
  39.             public static void main(String [] args)
  40.             {
  41.                 if(args == null || args.length == 0)
  42.                 {
  43.                     System.out.println("no args");
  44.                     System.exit(-1);
  45.                 }
  46.                 genCallLog(args[0]);
  47.  
  48.             }
  49.  
  50.             /**
  51.              * 生成通话日志
  52.              */
  53.             private static void genCallLog(String logFilePath) {
  54.  
  55.                 try {
  56.                     //文件写入器
  57.                     FileWriter fw = new FileWriter(logFilePath, true);
  58.                     Random random = new Random();
  59.  
  60.                     while (true) {
  61.  
  62.                         //主叫
  63.                         String caller = phoneNumbers.get(random.nextInt(callers.size()));
  64.                         String callerName = callers.get(caller);
  65.                         //被叫 （！= 主叫）
  66.                         String callee = phoneNumbers.get(random.nextInt(callers.size()));
  67.                         while (callee.equals(caller)) {
  68.                             callee = phoneNumbers.get(random.nextInt(callers.size()));
  69.                         }
  70.                         String calleeName = callers.get(callee);
  71.                         //通话时长(<10min)
  72.                         int duration = random.nextInt(60 * 10) + 1;
  73.                         DecimalFormat df = new DecimalFormat();
  74.                         df.applyPattern("000");
  75.                         String dur = df.format(duration);
  76.  
  77.                         //通话时间timeStr
  78.                         int year = 2018;
  79.                         int month = random.nextInt(12);
  80.                         int day = random.nextInt(29) + 1;
  81.                         int hour = random.nextInt(24);
  82.                         int min = random.nextInt(60);
  83.                         int sec = random.nextInt(60);
  84.  
  85.                         Calendar calendar = Calendar.getInstance();
  86.                         calendar.set(year,month,day,hour,min,sec);
  87.  
  88.                         Date date = calendar.getTime();
  89.                         //如果时间超过今天就重新qushijian取时间.
  90.                         Date now = new Date();
  91.                         if (date.compareTo(now) > 0) {
  92.                             continue ;
  93.                         }
  94.  
  95.                         SimpleDateFormat dfs = new SimpleDateFormat();
  96.                         dfs.applyPattern("yyyy/MM/dd HH:mm:ss");
  97.                         String timeStr = dfs.format(date);
  98.  
  99.                         //通话日志
  100.                         //String callLog = caller + "," + callerName + "," + callee + "," + calleeName + "," + timeStr + "," + dur;
  101.                         String callLog = caller + ","  + callee  + "," + timeStr + "," + dur;
  102.                         fw.write(callLog+ "\r\n");
  103.                         fw.flush();
  104.                         Thread.sleep(200);
  105.                     }
  106.                 } catch (Exception e) {
  107.                     e.printStackTrace();
  108.                 }
  109.  
  110.             }
  111.         }
  1.     4.打成jar包，扔到Linux上执行s100/s200
  2.         b.ubuntu上创建目录
  3.             $> mkdir /home/ubuntu/calllog
  4.  
  5.         a.执行命令
  6.             cmd> java -cp CallLogGenModel-1.0-SNAPSHOT.jar calllog.gen.main.App d:\\calllog\\calllog.log
  7.             $>   java -cp /share/calllog/CallLogGenModel-1.0-SNAPSHOT.jar calllog.gen.main.App /home/ubuntu/calllog/calllog.log
  8.  
  9.         c.编写快捷脚本 ~/calllog/calllog.sh
  10.             #!/bin/bash
  11.             java -cp /share/calllog/CallLogGenModel-1.0-SNAPSHOT.jar calllog.gen.main.App /home/ubuntu/calllog/calllog.log
  12.  
  13.         d.修改calllog.sh权限
  14.             $calllog> chmod 777 calllog.sh
  15.  
  16.         e.执行calllog.sh脚本
  17.             $calllog> ./calllog.sh
  18.  
  19.  
  20. 四、启动s100 s200 的flume,开始实时收集日志calllog.log [s100 s200]
  21. ---------------------------------------------------------
  22.     1.编写flume配置文件[flume/conf/calllog.conf]
  23.         
  1. a1.sources = r1
  2.         a1.sinks = k1
  3.         a1.channels = c1
  4.  
  5.         a1.sources.r1.type=exec
  6.         # -c +0 如果从头开始收集  -F:持续收集后续数据,否则进程停止。
  7.         a1.sources.r1.command=tail -F -c +0 /home/ubuntu/calllog/calllog.log
  8.  
  9.         a1.channels.c1.type=memory
  10.  
  11.         a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  12.         a1.sinks.k1.kafka.topic = calllog
  13.         a1.sinks.k1.kafka.bootstrap.servers = s200:9092 s300:9092 s400:9092
  14.         a1.sinks.k1.kafka.flumeBatchSize = 20
  15.         a1.sinks.k1.kafka.producer.acks = 1
  16.  
  17.         a1.sources.r1.channels = c1
  18.         a1.sinks.k1.channel = c1
  1.     2.在s100和s200上启动flume，开始收集日志
  2.         $s100> flume-ng agent -f /soft/flume/conf/calllog.conf -n a1 &
  3.         $s200> flume-ng agent -f /soft/flume/conf/calllog.conf -n a1 &
  4.  
  5.  
  6. 五、启动kafka集群
  7. --------------------------------------------------
  8.     1.启动zk集群[s100 s200 s300]
  9.         $> zkServer.sh start
  10.         $> xcall.sh jps
  11.  
  12.     2.启动kafka集群[s200 s300 s400]
  13.         $> /soft/kafka/bin/kafka-server-start.sh -daemon /soft/kafka/config/server.properties
  14.         $> netstat -ano | grep 9092
  15.  
  16.     3.创建kafka主题
  17.         $> kafka-topics.sh --create --zookeeper s100:2181 --replication-factor 3 --partitions 4 --topic calllog
  18.         $> kafka-topics.sh --list --zookeeper s100:2181
  19.  
  20.     4.在s300上开启kafka控制台消费者,消费flume收集的calllog主题,用于测试.
  21.         $s300> kafka-console-consumer.sh --zookeeper s100:2181 --topic calllog
  22.  
  23.     5.在s100和s200上 开启日志生成app,查看s300控制台输出情况
  24.         $s100> ~/calllog/calllog.sh
  25.         $s200> ~/calllog/calllog.sh
  26.  
  27.  
  28. 六、编写真正的kafka消费者HBase -- 从kafka提取消息，存放到hbase中
  29. ---------------------------------------------------------------------
  30.     1.启动hadoop集群[s100 s500 / s200 s300 s400], 完全分布式 + HA
  31.         a.$s100> start-all.sh
  32.  
  33.         b.$s100> xcall.sh jps
  34.             6656 Jps
  35.             6353 ResourceManager
  36.             6261 DFSZKFailoverController
  37.             3317 QuorumPeerMain
  38.             5818 NameNode
  39.  
  40.             ----xcall : jps from s200 ----
  41.             6224 DataNode
  42.             6721 NodeManager
  43.             7025 Jps
  44.             6465 JournalNode
  45.             3847 QuorumPeerMain
  46.             4335 Kafka
  47.  
  48.             ----xcall : jps from s300 ----
  49.             6088 NodeManager
  50.             6409 Jps
  51.             4330 Kafka
  52.             5595 DataNode
  53.             5836 JournalNode
  54.             3612 QuorumPeerMain
  55.  
  56.             ----xcall : jps from s400 ----
  57.             4242 Kafka
  58.             5241 DataNode
  59.             5738 NodeManager
  60.             5482 JournalNode
  61.             6059 Jps
  62.  
  63.             ----xcall : jps from s500 ----
  64.             5317 Jps
  65.             5064 DFSZKFailoverController
  66.             4826 NameNode
  67.  
  68.         c.查看webui
  69.             http://s100:50070
  70.  
  71.     2.启动hbase集群[s100 s500/ s200 s300 s400]
  72.         a.在s100上启动集群
  73.             $s100>start-hbase.sh
  74.  
  75.         b.在s500上启动备份master节点
  76.             $s500> hbase-daemon.sh start master
  77.  
  78.         c.查看webui
  79.             http://s100:16010
  80.  
  81.     3.创建hbase名字空间 + 表
  82.         a.进入hbase终端
  83.             $s100> hbase shell
  84.  
  85.         b.创建名字空间和表
  86.             $s100> create_namespace 'call'
  87.             $s100> create 'call:calllogs','f1'
  88.  
  89.     4.编程实现 -- 创建kafka消费者，订阅calllog主题
  90.         a.创建模块CalllogCustomerModel，添加maven支持
  91.  
  92.         b.添加maven依赖
  93.            
  1.  <?xml version="1.0" encoding="UTF-8"?>
  2.             <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.                      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.                      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.                 <modelVersion>4.0.0</modelVersion>
  6.  
  7.                 <groupId>groupId</groupId>
  8.                 <artifactId>CalllogCustomerModel</artifactId>
  9.                 <version>1.0-SNAPSHOT</version>
  10.  
  11.                 <dependencies>
  12.  
  13.                     <dependency>
  14.                         <groupId>org.apache.kafka</groupId>
  15.                         <artifactId>kafka_2.11</artifactId>
  16.                         <version>0.10.0.1</version>
  17.                     </dependency>
  18.                     <dependency>
  19.                         <groupId>org.apache.hbase</groupId>
  20.                         <artifactId>hbase-client</artifactId>
  21.                         <version>1.2.4</version>
  22.                     </dependency>
  23.  
  24.                 </dependencies>
  25.  
  26.             </project>
  1.         c.创建包
  2.             calllog.kafka.hbase.customer
  3.  
  4.         d.编写属性文件[resources/kafka.properties]
  5.             zookeeper.connect=s100:2181,s200:2181,s300:2181
  6.             group.id=calllog
  7.             zookeeper.session.timeout.ms=500
  8.             zookeeper.sync.time.ms=250
  9.             auto.commit.interval.ms=1000
  10.             #从头消费
  11.             auto.offset.reset=smallest
  12.             #主题
  13.             topic=calllog
  14.             #表名
  15.             table.name=call:calllogs
  16.             #分区数
  17.             partition.number=100
  18.             #主叫标记
  19.             caller.flag=0
  20.             #hash区域的模式
  21.             hashcode.pattern=00
  22.  
  23.         e.拷贝hbase-site.xml配置文件到resources目录下
  24.            
  1.  <?xml version="1.0"?>
  2.             <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
  3.  
  4.             <configuration>
  5.  
  6.                 <!-- 使用完全分布式 -->
  7.                 <property>
  8.                     <name>hbase.cluster.distributed</name>
  9.                     <value>true</value>
  10.                 </property>
  11.  
  12.                 <!-- 指定hbase数据在hdfs上的存放路径 -->
  13.                 <property>
  14.                     <name>hbase.rootdir</name>
  15.                     <value>hdfs://mycluster/hbase</value>
  16.                 </property>
  17.                 <!-- 配置zk地址 -->
  18.                 <property>
  19.                     <name>hbase.zookeeper.quorum</name>
  20.                     <value>192.168.43.131:2181,192.168.43.132:2181,192.168.43.133:2181</value>
  21.                 </property>
  22.                 <!-- zk的本地目录 -->
  23.                 <property>
  24.                     <name>hbase.zookeeper.property.dataDir</name>
  25.                     <value>/home/ubuntu/zookeeper</value>
  26.                 </property>
  27.             </configuration>
  1.         f.拷贝hdfs-site.xml到resources目录下
  2.             
  1. <?xml version="1.0" encoding="UTF-8"?>
  2.               <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
  3.  
  4.  
  5.               <configuration>
  6.                 <property>
  7.                     <name>dfs.replication</name>
  8.                     <value>3</value>
  9.                 </property>
  10.  
  11.                    <property>
  12.                      <name>dfs.ha.automatic-failover.enabled</name>
  13.                      <value>true</value>
  14.                    </property>
  15.  
  16.                 <property>
  17.                         <name>dfs.hosts</name>
  18.                         <value>/soft/hadoop/etc/dfs-hosts-include.conf</value>
  19.                     </property>
  20.  
  21.                 <property>
  22.                         <name>dfs.hosts.exclude</name>
  23.                         <value>/soft/hadoop/etc/dfs-hosts-exclude.conf</value>
  24.                     </property>
  25.  
  26.  
  27.               <!-- set ha  -->
  28.  
  29.  
  30.                 <property>
  31.                     <name>dfs.nameservices</name>
  32.                     <value>mycluster</value>
  33.                 </property>
  34.  
  35.                 <!-- myucluster下的名称节点两个id -->
  36.                 <!-- 注意：目前仅允许2个名称节点 -->
  37.                 <property>
  38.                     <name>dfs.ha.namenodes.mycluster</name>
  39.                     <value>nn1,nn2</value>
  40.                 </property>
  41.  
  42.                 <property>
  43.                     <name>dfs.namenode.rpc-address.mycluster.nn1</name>
  44.                     <value>s100:8020</value>
  45.                 </property>
  46.                 <property>
  47.                     <name>dfs.namenode.rpc-address.mycluster.nn2</name>
  48.                     <value>s500:8020</value>
  49.                 </property>
  50.                 <property>
  51.                     <name>dfs.namenode.http-address.mycluster.nn1</name>
  52.                     <value>s100:50070</value>
  53.                 </property>
  54.                 <property>
  55.                     <name>dfs.namenode.http-address.mycluster.nn2</name>
  56.                     <value>s500:50070</value>
  57.                 </property>
  58.  
  59.                 <property>
  60.                     <name>dfs.namenode.shared.edits.dir</name>
  61.                     <value>qjournal://s200:8485;s300:8485;s400:8485/mycluster</value>
  62.                 </property>
  63.  
  64.                 <property>
  65.                     <name>dfs.client.failover.proxy.provider.mycluster</name>
  66.                     <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
  67.                 </property>
  68.  
  69.                 <property>
  70.                     <name>dfs.ha.fencing.methods</name>
  71.                     <value>
  72.                         sshfence
  73.                         shell(/bin/true)
  74.                     </value>
  75.                 </property>
  76.                 <property>
  77.                     <name>dfs.ha.fencing.ssh.private-key-files</name>
  78.                     <value>/home/ubuntu/.ssh/id_rsa</value>
  79.                 </property>
  80.  
  81.                 <property>
  82.                     <name>dfs.journalnode.edits.dir</name>
  83.                     <value>/home/ubuntu/hadoop/journal</value>
  84.                 </property>
  85.  
  86.               <!-- set ha over  -->
  87.  
  88.  
  89.               </configuration>
  1.         g.编写工具类PropertiesUtil -- 外部加载prop
  2.         ---------------------------------------------------
  3.            
  1. package calllog.kafka.hbase.customer;
  2.            import java.io.IOException;
  3.            import java.io.InputStream;
  4.            import java.util.Properties;
  5.  
  6.            public class PropertiesUtil {
  7.  
  8.                public static Properties props;
  9.  
  10.                static {
  11.                    try {
  12.                        //外部加载属性文件props
  13.                        InputStream is = ClassLoader.getSystemResourceAsStream("kafka.properties");
  14.                        props = new Properties();
  15.                        props.load(is);
  16.                        is.close();
  17.                    } catch (IOException e) {
  18.                        e.printStackTrace();
  19.                    }
  20.                }
  21.  
  22.                /**
  23.                 * 获取属性
  24.                 */
  25.                public static String getPorp(String key)
  26.                {
  27.                    return props.getProperty(key);
  28.                }
  29.            }
  1.         h.编写类HbaseDao类 -- Hbase的访问数据对象，通过dao访问hbase
  2.             1)设rowkey:常用的主要指标，全部编写进来，而且要保证定长
  3.               区域号[00-99] , 1_id[主号码] , time , 标识[0/1  主叫/背叫] , 2_id[从属号码] , 时长
  4.               区域号[00-99] = (1_id[后四位] + time[yyyyMM]).hash()  %   100[区域数]
  5.  
  6.             2)代码
  7.             --------------------------------------
  8.             
  1. package calllog.kafka.hbase.customer;
  2.  
  3.             import org.apache.hadoop.conf.Configuration;
  4.             import org.apache.hadoop.hbase.HBaseConfiguration;
  5.             import org.apache.hadoop.hbase.TableName;
  6.             import org.apache.hadoop.hbase.client.Connection;
  7.             import org.apache.hadoop.hbase.client.ConnectionFactory;
  8.             import org.apache.hadoop.hbase.client.Put;
  9.             import org.apache.hadoop.hbase.client.Table;
  10.             import org.apache.hadoop.hbase.util.Bytes;
  11.  
  12.             import java.io.IOException;
  13.             import java.text.DecimalFormat;
  14.  
  15.             /**
  16.              * hbase的数据访问对象
  17.              */
  18.             public class HbaseDao {
  19.  
  20.                 private Table table = null;
  21.                 private DecimalFormat df = new DecimalFormat();
  22.                 //设计rowkey的分区标识
  23.                 private int partitions;
  24.                 //0 -- 主叫  1 -- 被叫
  25.                 private String flag;
  26.  
  27.                 public HbaseDao() {
  28.  
  29.                     try {
  30.                         //获取配置文件
  31.                         Configuration conf = HBaseConfiguration.create();
  32.                         //工厂类创建连接
  33.                         Connection conn = ConnectionFactory.createConnection(conf);
  34.                         //get table
  35.                         TableName tbName = TableName.valueOf(PropertiesUtil.getPorp("table.name"));
  36.                         table = conn.getTable(tbName);
  37.                         df.applyPattern(PropertiesUtil.getPorp("hashcode.pattern"));
  38.                         partitions = Integer.parseInt(PropertiesUtil.getPorp("partition.number"));
  39.                         flag = PropertiesUtil.getPorp("caller.flag");
  40.                     } catch (IOException e) {
  41.                         e.printStackTrace();
  42.                     }
  43.                 }
  44.  
  45.  
  46.                 /**
  47.                  * 向hbase中put数据
  48.                  */
  49.                 public void put(String log)
  50.                 {
  51.                     if(log == null || log.equals(""))
  52.                     {
  53.                         return;
  54.                     }
  55.  
  56.                     try {
  57.                         //设计rowkey
  58.                         String rowKey = "";
  59.                         //解析日志
  60.                         String [] strs = log.split(",");
  61.                         if(strs != null && strs.length == 4)
  62.                         {
  63.                             String caller = strs[0];
  64.                             String callee = strs[1];
  65.                             String time = strs[2];
  66.                             String duration = strs[3];
  67.                             //计算区域号
  68.                             String hash = getRegionNumber(caller, time);
  69.                             rowKey = getRowkey(hash,caller,flag,time,callee,duration);
  70.  
  71.                             //开始put
  72.                             Put p = new Put(Bytes.toBytes(rowKey));
  73.                             p.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("caller"),Bytes.toBytes(caller));
  74.                             p.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("callee"),Bytes.toBytes(callee));
  75.                             p.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("callTime"),Bytes.toBytes(time));
  76.                             p.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("callDuration"),Bytes.toBytes(duration));
  77.                             table.put(p);
  78.                         }
  79.                     } catch (IOException e) {
  80.                         e.printStackTrace();
  81.                     }
  82.                 }
  83.  
  84.                 /**
  85.                  * 获取区域号码 -- Rowkey设计用
  86.                  * @return
  87.                  */
  88.                 public String getRegionNumber(String caller, String calltime)
  89.                 {
  90.                     //取得电话号码的后四位
  91.                     String last4Code = caller.substring(caller.length() - 4);
  92.                     //取得通话时间的年月
  93.                     String month = calltime.substring(0, 6);
  94.                     int hash = (Integer.parseInt(last4Code) ^ Integer.parseInt(month)) % partitions;
  95.                     return df.format(hash);
  96.                 }
  97.  
  98.                 /**
  99.                  * 获取rowkey
  100.                  */
  101.                 public String getRowkey(String hash, String caller,String time,String flag, String callee,String dur)
  102.                 {
  103.                     return hash + "," + caller + "," + time + "," + flag + "," + callee + "," + dur;
  104.                 }
  105.  
  106.             }
  1.         i.编写主类 -- HbaseCustomer
  2.         ---------------------------------------------
  3.          
  1. package calllog.kafka.hbase.customer;
  2.  
  3.          import kafka.consumer.Consumer;
  4.          import kafka.consumer.ConsumerConfig;
  5.          import kafka.message.MessageAndMetadata;
  6.  
  7.          import java.io.IOException;
  8.          import java.io.InputStream;
  9.          import java.util.HashMap;
  10.          import java.util.List;
  11.          import java.util.Map;
  12.          import java.util.Properties;
  13.          import kafka.consumer.ConsumerIterator;
  14.          import kafka.consumer.KafkaStream;
  15.          import kafka.javaapi.consumer.ConsumerConnector;
  16.  
  17.          import java.util.Properties;
  18.  
  19.          /**
  20.           * hbase消费者，从kafka获取日志信息，存储到hbase中
  21.           */
  22.          public class HbaseCustomer {
  23.  
  24.  
  25.              public static void main(String [] args)
  26.              {
  27.                  //hbasedao
  28.                  HbaseDao dao = new HbaseDao();
  29.                  //创建消费者配置文件
  30.                  ConsumerConfig config = new ConsumerConfig(PropertiesUtil.props);
  31.                  //创建消费者
  32.                  ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(PropertiesUtil.props));
  33.                  //绑定主题
  34.                  String topic = PropertiesUtil.getPorp("topic");
  35.  
  36.                  Map<String, Integer> map = new HashMap<String, Integer>();
  37.                  map.put(topic, new Integer(1));
  38.  
  39.                  //开始消费
  40.                  Map<String, List<KafkaStream<byte[], byte[]>>> kafkaMsg = consumer.createMessageStreams(map);
  41.                  List<KafkaStream<byte[], byte[]>> msgList = kafkaMsg.get(topic);
  42.  
  43.                  String kafka_hbaseMsg = "";
  44.                  for(KafkaStream<byte[],byte[]> msg : msgList)
  45.                  {
  46.                      ConsumerIterator<byte[],byte[]> mm = msg.iterator();
  47.                      while (mm.hasNext()) {
  48.                          MessageAndMetadata<byte[], byte[]> next = mm.next();
  49.                          byte [] m = next.message();
  50.                          //获取消息
  51.                          kafka_hbaseMsg = new String(m);
  52.                          //写入hbase
  53.                          dao.put(kafka_hbaseMsg);
  54.                      }
  55.                  }
  56.              }
  57.          }
  1.         j.使用idea进行关联jar打包
  2.             File --> project strructure --> artifacts --> ...
  3.  
  4.         l.在window上执行，测试
  5.             cmd> java -cp CalllogCustomerModel.jar calllog.kafka.hbase.customer.HbaseCustomer
  6.  
  7.         m.将jar包放到共享文件夹下，在ubuntu上执行程序，查看程序是否正确执行
  8.             $> java -cp CalllogCustomerModel.jar calllog.kafka.hbase.customer.HbaseCustomer
  9.  
  10.  
  11. 七、编写web程序，从hbase中提取数据，进行可视化展示
  12. --------------------------------------------------------
  13.     1.导入上次课程的SSM工程
  14.         File --> project struct --> Models --> + --> ssm.imi --> meven '+' 添加pom.xml
  15.  
  16.     2.进行一系列web设置：
  17.         a. setting --> Application Server --> 添加tomcat服务器
  18.  
  19.         b. File --> project structure --> Artifacs --> + 添加ssmweb模块 --> +添加 右侧支持依赖的jar包和依赖的外部配置文件
  20.  
  21.         c. Run --> edit configuarations --> 添加tomcat local app --> deployment + --> 添加自己的web模块(ssm) --> 部署热更新
  22.  
  23.  
  24.     3.运行，输入网址 http://localhost:8080/user/findall?pn=1,测试程序
  25.  
  26.     4.在domain包中添加calllog类
  27.     ----------------------------------------------
  28.        
  1.  package com.it18zhang.ssm.domain;
  2.  
  3.         /**
  4.          * calllog的domain类 -- 标准javabean
  5.          */
  6.         public class Calllog {
  7.  
  8.             private String caller;
  9.             private String callee;
  10.             private String callTime;
  11.             private String callDuration;
  12.  
  13.             public String getCaller() {
  14.                 return caller;
  15.             }
  16.  
  17.             public void setCaller(String caller) {
  18.                 this.caller = caller;
  19.             }
  20.  
  21.             public String getCallee() {
  22.                 return callee;
  23.             }
  24.  
  25.             public void setCallee(String callee) {
  26.                 this.callee = callee;
  27.             }
  28.  
  29.             public String getCallTime() {
  30.                 return callTime;
  31.             }
  32.  
  33.             public void setCallTime(String callTime) {
  34.                 this.callTime = callTime;
  35.             }
  36.  
  37.             public String getCallDuration() {
  38.                 return callDuration;
  39.             }
  40.  
  41.             public void setCallDuration(String callDuration) {
  42.                 this.callDuration = callDuration;
  43.             }
  44.         }
  1.     5.添加calllog service接口CalllogService.interface
  2.     ------------------------------------------------------------
  3.        
  1.  package com.it18zhang.ssm.service;
  2.  
  3.         import Calllog;
  4.  
  5.         import java.util.List;
  6.  
  7.         /**
  8.          * Calllog的服务类 -- 用于定制与服务器交互的规则
  9.          */
  10.         public interface CalllogService {
  11.  
  12.             //查询所有的calllog
  13.             public List<Calllog> findAll();
  14.  
  15.         }
  1.     6.添加CalllogService的实现类CalllogServiceImpl，用于与hbase进行交互
  2.     -------------------------------------------------------------------------
  3.         a.准备必要的配置文件，拷贝[hbase-site.xml / hdfs-site.xml]到resouces目录下
  4.  
  5.         b.添加maven依赖
  6.             <dependency>
  7.                 <groupId>org.apache.kafka</groupId>
  8.                 <artifactId>kafka_2.11</artifactId>
  9.                 <version>0.10.0.1</version>
  10.             </dependency>
  11.             <dependency>
  12.                 <groupId>org.apache.hbase</groupId>
  13.                 <artifactId>hbase-client</artifactId>
  14.                 <version>1.2.4</version>
  15.             </dependency>
  16.  
  17.         c.编写类CalllogServiceImpl
  18.         ------------------------------------------
  19.            
  1.  package com.it18zhang.ssm.service.impl;
  2.  
  3.             import com.it18zhang.ssm.domain.Calllog;
  4.             import com.it18zhang.ssm.service.CalllogService;
  5.             import org.apache.hadoop.conf.Configuration;
  6.             import org.apache.hadoop.hbase.HBaseConfiguration;
  7.             import org.apache.hadoop.hbase.TableName;
  8.             import org.apache.hadoop.hbase.client.*;
  9.             import org.apache.hadoop.hbase.util.Bytes;
  10.             import org.springframework.stereotype.Service;
  11.  
  12.             import java.io.IOException;
  13.             import java.util.*;
  14.  
  15.             /**
  16.              * CalllogService的实现类
  17.              */
  18.             @Service("calllogService")
  19.             public class CalllogServiceImpl implements CalllogService {
  20.  
  21.                 private Table table;
  22.                 public CalllogServiceImpl()
  23.                 {
  24.                     try {
  25.                         //获取配置文件
  26.                         Configuration conf = HBaseConfiguration.create();
  27.                         //工厂类创建连接
  28.                         Connection conn = ConnectionFactory.createConnection(conf);
  29.                         //get table
  30.                         TableName tbName = TableName.valueOf("call:calllogs");
  31.                         table = conn.getTable(tbName);
  32.  
  33.                     } catch (IOException e) {
  34.                         e.printStackTrace();
  35.                     }
  36.                 }
  37.  
  38.                 /**
  39.                  * 查询所有的calllog
  40.                  * 全表扫描
  41.                  * @return
  42.                  */
  43.                 public List<Calllog> findAll() {
  44.                     List<Calllog> list = new ArrayList<Calllog>();
  45.                     try {
  46.                         //扫描
  47.                         Scan scan = new Scan();
  48.                         ResultScanner rs = table.getScanner(scan);
  49.                         Iterator<Result> it = rs.iterator();
  50.                         byte[] famliy = Bytes.toBytes("f1");
  51.                         byte[] callerf = Bytes.toBytes("caller");
  52.                         byte[] calleef = Bytes.toBytes("callee");
  53.                         byte[] callTimef = Bytes.toBytes("callTime");
  54.                         byte[] callDurationf = Bytes.toBytes("callDuration");
  55.                         Calllog calllog = null;
  56.  
  57.                         while (it.hasNext()) {
  58.  
  59.                             Result next = it.next();
  60.                             String caller = Bytes.toString(next.getValue(famliy, callerf));
  61.                             String callee = Bytes.toString(next.getValue(famliy, calleef));
  62.                             String callTime = Bytes.toString(next.getValue(famliy, callTimef));
  63.                             String callDuration = Bytes.toString(next.getValue(famliy, callDurationf));
  64.  
  65.                             calllog = new Calllog();
  66.                             calllog.setCaller(caller);
  67.                             calllog.setCallee(callee);
  68.                             calllog.setCallTime(callTime);
  69.                             calllog.setCallDuration(callDuration);
  70.  
  71.                             list.add(calllog);
  72.                         }
  73.                     } catch (Exception e) {
  74.                         e.printStackTrace();
  75.                     }
  76.                     return list;
  77.                 }
  78.             }
  1.         7.添加CalllogContorller -- 用于web界面显示
  2.         -------------------------------------------------------
  3.            
  1. package com.it18zhang.ssm.web.controller;
  2.  
  3.            import com.it18zhang.ssm.domain.Calllog;
  4.            import com.it18zhang.ssm.service.CalllogService;
  5.            import com.it18zhang.ssm.service.impl.CalllogServiceImpl;
  6.            import org.springframework.stereotype.Controller;
  7.            import org.springframework.ui.Model;
  8.            import org.springframework.web.bind.annotation.RequestMapping;
  9.  
  10.            import javax.annotation.Resource;
  11.            import java.util.List;
  12.  
  13.            @Controller
  14.            public class CalllogController {
  15.  
  16.                @Resource(name="calllogService")
  17.                private CalllogService cs;
  18.  
  19.                @RequestMapping("calllog/findAll")
  20.                public String findAll(Model model)
  21.                {
  22.                    List<Calllog> list = cs.findAll();
  23.                    model.addAttribute("calllogs", list);
  24.                    return "calllog/calllogList";
  25.                }
  26.            }
        8.添加jsp页面calllog/calllogList.jsp     

  1.   <%@ page contentType="text/html;charset=UTF-8" language="java" %>
  2.             <%@ taglib uri="http://java.sun.com/jsp/jstl/core" prefix="c" %>
  3.             <html>
  4.             <head>
  5.                 <title>通话记录</title>
  6.                 <link rel="stylesheet" type="text/css" href="../css/my.css">
  7.             </head>
  8.             <body>
  9.             <table id="t1" border="1px" class="t-1">
  10.                 <tr>
  11.                     <td>主叫</td>
  12.                     <td>被叫</td>
  13.                     <td>通话时间</td>
  14.                     <td>通话时长</td>
  15.                 </tr>
  16.                 <c:forEach items="${calllogs}" var="u">
  17.                     <tr>
  18.                         <td><c:out value="${u.caller}"/></td>
  19.                         <td><c:out value="${u.callee}"/></td>
  20.                         <td><c:out value="${u.callTime}"/></td>
  21.                         <td><c:out value="${u.callDuration}"/></td>
  22.                     </tr>
  23.                 </c:forEach>
  24.                 <tr>
  25.                     <td colspan="5" style="text-align: right">
  26.                     </td>
  27.                 </tr>
  28.             </table>
  29.             </body>
  30.             </html>

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/301026
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号