devbox
菜鸟追梦旅行
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

大数据案例--网站流量项目（中）_大数据项目开发案例

作者：菜鸟追梦旅行 | 2024-05-31 06:05:47

赞

踩

大数据项目开发案例

目录 

一、Hive做离线批处理

1、实现步骤

①、启动hadoop，启动hive

②、在hive下创建weblog库，并使用

③、 创建外部表管理数据

④、为总表添加当天分区数据

⑤、建立数据清洗表，用于清洗出业务所需的字段。

⑥、业务处理

⑦、创建业务表并插入数据

⑧、从清洗表查询得到当天的统计指标，插入到业务表中

⑨、利用Sqoop工具从HDFS上将数据导入到Mysql数据库中

二、Hive的占位符与文件调用

1、概述

2、Hive文件的调用

3、Hive占位符的使用

4、结合业务实现

5、Linux Crontab定时任务

三、实时业务系统搭建

1、Flume与Kafka的连通

四、实时流开发环境搭建

1、Spark与HBase整合基础

2、实时流业务处理

一、Hive做离线批处理

1、实现步骤

①、启动hadoop，启动hive

进入hive的bin目录（以后台方式启动）

nohup hive --service metastore &

nohup hive --service hiveserver2 &

sh hive

②、在hive下创建weblog库，并使用

create database weblog;
use weblog

③、 创建外部表管理数据

建立总表，用于管理所有的字段数据。

总表特点：管理所有字段，外部表，分区表

hdfs上的数据：

建表语句：

create external table flux (url string,urlname string,title string,chset string,scr string,col string,lg string,je string,ec string,fv string,cn string,ref string,uagent string,stat_uv string,stat_ss string,cip string) PARTITIONED BY (reporttime string) row format delimited fields terminated by '|' location '/weblog';

 

④、为总表添加当天分区数据

1、msck repair table flux;

2、alter table flux add partition(reporttime='2022-04-20') location '/weblog/reporttime=2022-04-20';

⑤、建立数据清洗表，用于清洗出业务所需的字段。

dataclear   指定的分割符 :   |

去除多余字段，只保留需要的字段，并将会话信息拆开保存

所需要的字段为：

reporttime、url、urlname、uvid、ssid、sscount、sstime、cip

create table dataclear(reportTime string,url string,urlname string,uvid string,ssid string,sscount string,sstime string,cip string)row format delimited fields terminated by '|';

 从总表中查询出当天的对应的字段插入到清洗表中

  1. insert overwrite table dataclear
  2. select reporttime,url,urlname,stat_uv,split(stat_ss,"_")[0],split(stat_ss,"_")[1],split(stat_ss,"_")[2],cip  from flux;

⑥、业务处理

1、pv

select count(*) as pv from dataclear where reportTime = '2022-04-20';

2、uv

uv - 独立访客数 - 一天之内所有的访客的数量 - 一天之内uvid去重后的总数

select count(distinct uvid) as uv from dataclear where reportTime = '2022-04-20';

3、vv

vv - 独立会话数 - 一天之内所有的会话的数量 - 一天之内ssid去重后的总数

select count(distinct ssid) as vv from dataclear where reportTime = '2022-04-20';

4、br

br - 跳出率 - 一天内跳出的会话总数/会话总数

select round(br_taba.a/br_tabb.b,4)as br from (select count(*) as a from (select ssid from dataclear where reportTime='2022-04-20' group by ssid having count(ssid) = 1) as br_tab) as br_taba,(select count(distinct ssid) as b from dataclear where reportTime='2022-04-20') as br_tabb;

这段sql就是对会话id分组，然后求出会话id为1的个数，这个就是跳出会话

5、newip

newip - 新增ip总数 - 一天内所有ip去重后在历史数据中从未出现过的数量

select count(distinct dataclear.cip) from dataclear where dataclear.reportTime = '2022-04-20' and cip not in (select dc2.cip from dataclear as dc2 where dc2.reportTime < '2022-04-20');

6、newcust

newcust - 新增客户数 - 一天内所有的uvid去重后在历史数据中从未出现过的总数

select count(distinct dataclear.uvid) from dataclear where dataclear.reportTime='2021-05-10'

and uvid not in

(select dc2.uvid from dataclear as dc2 where dc2.reportTime < '2021-05-10');

7、avgtime

avgtime - 平均访问时常 - 一天内所有会话的访问时常的平均值

注： 一个会话的时长 = 会话中所有访问的时间的最大值 - 会话中所有访问时间的最小值

select avg(atTab.usetime) as avgtime from(select max(sstime) - min(sstime) as usetime from dataclear where reportTime='2022-04-20' group by ssid) as atTab;

8、avgdeep

avgdeep - 平均访问深度 - 一天内所有会话访问深度的平均值

一个会话的访问深度=一个会话访问的所有url去重后的个数

比如会话①：url http://demo/a.jsp     http://demo/b.jsp   http://demo/a.jsp 则访问深度是2

select round(avg(adTab.deep),4) as avgdeep from (select count(distinct urlname) as deep from dataclear where reportTime='2022-04-20' group by ssid) as adTab;

⑦、创建业务表并插入数据

create table tongji(reportTime string,pv int,uv int,vv int, br double,newip int, newcust int, avgtime double,avgdeep double) row format delimited fields terminated by '|';

⑧、从清洗表查询得到当天的统计指标，插入到业务表中

  1. insert overwrite table tongji  select '2022-04-20',tab1.pv,tab2.uv,tab3.vv,tab4.br,tab5.newip,tab6.newcust,tab7.avgtime,tab8.avgdeep from
  2.  (select count(*) as pv from dataclear where reportTime = '2022-04-20') as tab1,
  3. (select count(distinct uvid) as uv from dataclear where reportTime = '2022-04-20') as tab2,
  4. (select count(distinct ssid) as vv from dataclear where reportTime = '2022-04-20') as tab3,
  5. (select round(br_taba.a/br_tabb.b,4)as br from (select count(*) as a from (select ssid from dataclear where reportTime='2022-04-20' group by ssid having count(ssid) = 1) as br_tab) as br_taba,
  6. (select count(distinct ssid) as b from dataclear where reportTime='2022-04-20') as br_tabb) as tab4,
  7. (select count(distinct dataclear.cip) as newip from dataclear where dataclear.reportTime = '2022-04-20' and cip not in (select dc2.cip from dataclear as dc2 where dc2.reportTime < '2022-04-20')) as tab5,
  8. (select count(distinct dataclear.uvid) as newcust from dataclear where dataclear.reportTime='2022-04-20' and uvid not in (select dc2.uvid from dataclear as dc2 where dc2.reportTime < '2022-04-20')) as tab6,
  9. (select round(avg(atTab.usetime),4) as avgtime from (select max(sstime) - min(sstime) as usetime from dataclear where reportTime='2022-04-20' group by ssid) as atTab) as tab7,
  10. (select round(avg(deep),4) as avgdeep from (select count(distinct urlname) as deep from dataclear where reportTime='2022-04-20' group by ssid) as adTab) as tab8;

⑨、利用Sqoop工具从HDFS上将数据导入到Mysql数据库中

进入mysql：

mysql -uroot -proot
创建和使用库：

create database weblog;
use weblog;

创建表：

create table tongji(reporttime varchar(40),pv int,uv int,vv int,br double,newip int,newcust int,avgtime double,avgdeep double);

使用Sqoop将数据导入到mysql中：

进入：

cd /home/software/sqoop-1.4.7/bin/
执行：

sh sqoop export --connect jdbc:mysql://hadoop01:3306/weblog --username root --password root --export-dir '/user/hive/warehouse/weblog.db/tongji' --table tongji -m 1 --fields-terminated-by '|'

 进入mysql 查询：

二、Hive的占位符与文件调用

1、概述

对于上面的工作，我们发现需要手动去写hql语句从而完成离线数据的ETL，但每天都手动来做显然是不合适的，所以可以利用hive的文件调用与占位符来解决这个问题。

2、Hive文件的调用

实现步骤：

①、编写一个文件，后缀名为.hive

比如我们现在我们创建一个01.hive文件

目的是在 hive的weblog数据库下，创建一个tb1表

  1. use  weblog;
  2. create table tb1 (id int,name string)；

②、进入hive安装目录的bin目录

执行： sh hive -f  01.hive

注：-f  参数后跟的是01.hive文件的路径

③、测试hive的表是否创建成功

 

3、Hive占位符的使用

我们现在想通过hive执行文件，将 "tb1"这个表删除，则我们可以这样做：

①、创建02.hive文件

  1. use  weblog;
  2. drop table ${tb_name};

②、在bin目录下，执行：

sh hive -f 02.hive -d tb_name="tb1"

4、结合业务实现

在hive最后插入数据时，涉及到一个日志的分区是以每天为单位，所以我们需要手动去写这个日期，比如 2022-04-20。我们可以这样做：

①、将hql语句里的日期相关的取值用占位符来表示，并写在weblog.hive文件里

  1. use weblog;
  2. insert overwrite table tongji  select ${reportTime},tab1.pv,tab2.uv,tab3.vv,tab4.br,tab5.newip,tab6.newcust,tab7.avgtime,tab8.avgdeep from (select count(*) as pv from dataclear where reportTime = ${reportTime}) as tab1,(select count(distinct uvid) as uv from dataclear where reportTime = ${reportTime}) as tab2,(select count(distinct ssid) as vv from dataclear where reportTime = ${reportTime}) as tab3,(select round(br_taba.a/br_tabb.b,4)as br from (select count(*) as a from (select ssid from dataclear where reportTime=${reportTime} group by ssid having count(ssid) = 1) as br_tab) as br_taba,(select count(distinct ssid) as b from dataclear where reportTime=${reportTime}) as br_tabb) as tab4,(select count(distinct dataclear.cip) as newip from dataclear where dataclear.reportTime = ${reportTime} and cip not in (select dc2.cip from dataclear as dc2 where dc2.reportTime < ${reportTime})) as tab5,(select count(distinct dataclear.uvid) as newcust from dataclear where dataclear.reportTime=${reportTime} and uvid not in (select dc2.uvid from dataclear as dc2 where dc2.reportTime < ${reportTime})) as tab6,(select round(avg(atTab.usetime),4) as avgtime from (select max(sstime) - min(sstime) as usetime from dataclear where reportTime=${reportTime} group by ssid) as atTab) as tab7,(select round(avg(deep),4) as avgdeep from (select count(distinct urlname) as deep from dataclear where reportTime=${reportTime} group by ssid) as adTab) as tab8;

②、在hive 的bin目录下执行：

sh hive -f  weblog.hive -d reportTime="2022-04-20"

对于日期，如果不想手写的话，可以通过linux的指令来获取：

 date "+%G-%m-%d"

所以我们可以这样来执行hive文件的调用：

sh hive -f weblog.hive -d reportTime=`date "+%G-%m-%d"`(注：是键盘右上方的反引号）

也可以写为：

sh hive -f weblog.hive -d reportTime=$(date "+%G-%m-%d")

5、Linux Crontab定时任务

在工作中需要数据库在每天零点自动备份所以需要建立一个定时任务。

crontab命令的功能是在一定的时间间隔调度一些命令的执行。

可以通过 crontab -e 进行定时任务的编辑

crontab文件格式：

*               *          *        *             *           command

minute   hour    day     month   week      command

分            时         天      月          星期        命令

示例：

 */1 * * * * rm -rf /home/software/1.txt

     每隔一分钟，删除指定目录的 1.txt文件

 对于上面的项目，我们可以这样写：

45 23 * * * ./home/software/hive-3.1.2/bin/hive -f /home/software/hive-3.1.2/bin/weblog.hive -d time=`date "+%G-%m
-%d"`

三、实时业务系统搭建

1、Flume与Kafka的连通

1.启动zk集群

2.启动kafka集群在其bin目录下执行

指令：sh kafka-server-start.sh ../config/server.properties

3.创建主题

查看主题：sh kafka-topics.sh --list --zookeeper hadoop01:2181

创建主题：sh kafka-topics.sh --create --zookeeper hadoop01:2181 --replication-factor 1 --partitions 1 --topic fluxdata

4.配置flume的data下的weblog.conf（自己建的）

  1. a1.sources=r1
  2. a1.channels=c1 c2
  3. a1.sinks=k1 k2
  4.  
  5. a1.sources.r1.type=avro
  6. a1.sources.r1.bind=0.0.0.0
  7. a1.sources.r1.port=44444
  8. a1.sources.r1.interceptors=i1
  9. a1.sources.r1.interceptors.i1.type=timestamp
  10.  
  11. a1.sinks.k1.type=hdfs
  12. a1.sinks.k1.hdfs.path=hdfs://192.168.186.128:9000/weblog/reportTime=%Y-%m-%d
  13. a1.sinks.k1.hdfs.fileType=DataStream
  14. a1.sinks.k1.hdfs.rollInterval=0
  15. a1.sinks.k1.hdfs.rollSize=0
  16. a1.sinks.k1.hdfs.rollCount=1000
  17.  
  18. a1.sinks.k2.type=org.apache.flume.sink.kafka.KafkaSink
  19. a1.sinks.k2.brokerList=hadoop01:9092,hadoop02:9092,hadoop03:9092
  20. a1.sinks.k2.topic=fluxdata
  21.  
  22. a1.channels.c1.type=memory
  23. a1.channels.c1.capacity=1000
  24. a1.channels.c1.transactionCapacity=100
  25.  
  26. a1.channels.c2.type=memory
  27. a1.channels.c2.capacity=1000
  28. a1.channels.c2.transactionCapacity=100
  29.  
  30. a1.sources.r1.channels=c1 c2
  31. a1.sinks.k1.channel=c1
  32. a1.sinks.k2.channel=c2

启动hadoop

在flume的data目录，执行下面语句启动flume：

../bin/flume-ng agent -n a1 -c ./ -f ./weblog.conf -Dflume.root.logger=INFO,console

5.启动tomcat，访问埋点服务器

6.测试kafka是否能够收到数据

        进入kafka的bin目录，启动kafka消费者线程：

 sh kafka-console-consumer.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --topic fluxdata --from-beginning

这时候我们访问页面：

四、实时流开发环境搭建

1、Spark与HBase整合基础

实现步骤：

1、启动IDEA

2、创建Maven工程，骨架选择quickstart

3、IDEA安装scala

4、为FluxStreamingServer工程添加scala sdk

这里如果spark如果是2版本，我们scala用scala2.11.7,稳定；如果是3版本，我们可以用scala2.12.X

5、创建一个scala目录，使其称为sources root

6、引入工程pom

  1. <?xml version="1.0" encoding="UTF-8"?>
  2.  
  3. <project xmlns="http://maven.apache.org/POM/4.0.0" 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>org.example</groupId>
  8.   <artifactId>FluxStreamingServer</artifactId>
  9.   <version>1.0-SNAPSHOT</version>
  10.  
  11.   <name>FluxStreamingServer</name>
  12.   <!-- FIXME change it to the project's website -->
  13.   <url>http://www.example.com</url>
  14.  
  15.   <properties>
  16.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  17.     <maven.compiler.source>1.7</maven.compiler.source>
  18.     <maven.compiler.target>1.7</maven.compiler.target>
  19.   </properties>
  20.  
  21.   <dependencies>
  22.     <dependency>
  23.       <groupId>junit</groupId>
  24.       <artifactId>junit</artifactId>
  25.       <version>4.11</version>
  26.       <scope>test</scope>
  27.     </dependency>
  28.  
  29.     <!--spark -->
  30.     <dependency>
  31.       <groupId>org.apache.spark</groupId>
  32.       <artifactId>spark-core_2.12</artifactId>
  33.       <version>3.1.2</version>
  34.     </dependency>
  35.  
  36.     <dependency>
  37.       <groupId>org.apache.spark</groupId>
  38.       <artifactId>spark-streaming_2.12</artifactId>
  39.       <version>3.1.1</version>
  40.     </dependency>
  41.  
  42.     <dependency>
  43.       <groupId>org.apache.spark</groupId>
  44.       <artifactId>spark-sql_2.12</artifactId>
  45.       <version>3.1.2</version>
  46.     </dependency>
  47.  
  48.     <dependency>
  49.       <groupId>org.apache.spark</groupId>
  50.       <artifactId>spark-mllib_2.12</artifactId>
  51.       <version>3.1.2</version>
  52.     </dependency>
  53.  
  54.     <!-- kafka -->
  55.     <dependency>
  56.       <groupId>org.apache.kafka</groupId>
  57.       <artifactId>kafka_2.12</artifactId>
  58.       <version>2.8.0</version>
  59.     </dependency>
  60.  
  61.     <dependency>
  62.       <groupId>org.apache.spark</groupId>
  63.       <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
  64.       <version>3.1.2</version>
  65.     </dependency>
  66.  
  67.     <!--HBase-->
  68.     <dependency>
  69.       <groupId>org.apache.hbase</groupId>
  70.       <artifactId>hbase</artifactId>
  71.       <version>2.4.2</version>
  72.       <type>pom</type>
  73.     </dependency>
  74.  
  75.     <dependency>
  76.       <groupId>org.apache.hbase</groupId>
  77.       <artifactId>hbase-client</artifactId>
  78.       <version>2.4.2</version>
  79.     </dependency>
  80.  
  81.  
  82.     <dependency>
  83.       <groupId>org.apache.hbase</groupId>
  84.       <artifactId>hbase-common</artifactId>
  85.       <version>2.4.2</version>
  86.     </dependency>
  87.  
  88.     <dependency>
  89.       <groupId>org.apache.hbase</groupId>
  90.       <artifactId>hbase-server</artifactId>
  91.       <version>2.4.2</version>
  92.     </dependency>
  93.  
  94.     <dependency>
  95.       <groupId>org.apache.hbase</groupId>
  96.       <artifactId>hbase-protocol</artifactId>
  97.       <version>2.4.2</version>
  98.     </dependency>
  99.  
  100.     <dependency>
  101.       <groupId>org.apache.hbase</groupId>
  102.       <artifactId>hbase-hadoop-compat</artifactId>
  103.       <version>2.4.2</version>
  104.     </dependency>
  105.  
  106.     <dependency>
  107.       <groupId>org.apache.hbase</groupId>
  108.       <artifactId>hbase-mapreduce</artifactId>
  109.       <version>2.4.2</version>
  110.     </dependency>
  111.  
  112.     <!--mysql-->
  113.     <dependency>
  114.       <groupId>mysql</groupId>
  115.       <artifactId>mysql-connector-java</artifactId>
  116.       <version>5.1.38</version>
  117.     </dependency>
  118.  
  119.     <dependency>
  120.       <groupId>com.mchange</groupId>
  121.       <artifactId>c3p0</artifactId>
  122.       <version>0.9.5.5</version>
  123.     </dependency>
  124.  
  125.  
  126.   </dependencies>
  127.  
  128.   <build>
  129.     <pluginManagement><!-- lock down plugins versions to avoid using Maven defaults (may be moved to parent pom) -->
  130.       <plugins>
  131.         <!-- clean lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#clean_Lifecycle -->
  132.         <plugin>
  133.           <artifactId>maven-clean-plugin</artifactId>
  134.           <version>3.1.0</version>
  135.         </plugin>
  136.         <!-- default lifecycle, jar packaging: see https://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_jar_packaging -->
  137.         <plugin>
  138.           <artifactId>maven-resources-plugin</artifactId>
  139.           <version>3.0.2</version>
  140.         </plugin>
  141.         <plugin>
  142.           <artifactId>maven-compiler-plugin</artifactId>
  143.           <version>3.8.0</version>
  144.         </plugin>
  145.         <plugin>
  146.           <artifactId>maven-surefire-plugin</artifactId>
  147.           <version>2.22.1</version>
  148.         </plugin>
  149.         <plugin>
  150.           <artifactId>maven-jar-plugin</artifactId>
  151.           <version>3.0.2</version>
  152.         </plugin>
  153.         <plugin>
  154.           <artifactId>maven-install-plugin</artifactId>
  155.           <version>2.5.2</version>
  156.         </plugin>
  157.         <plugin>
  158.           <artifactId>maven-deploy-plugin</artifactId>
  159.           <version>2.8.2</version>
  160.         </plugin>
  161.         <!-- site lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#site_Lifecycle -->
  162.         <plugin>
  163.           <artifactId>maven-site-plugin</artifactId>
  164.           <version>3.7.1</version>
  165.         </plugin>
  166.         <plugin>
  167.           <artifactId>maven-project-info-reports-plugin</artifactId>
  168.           <version>3.0.0</version>
  169.         </plugin>
  170.  
  171.         <plugin>
  172.           <artifactId>maven-jar-plugin</artifactId>
  173.           <version>3.0.2</version>
  174.           <configuration>
  175.             <archive>
  176.               <manifest>
  177.                 <addClasspath>true</addClasspath>
  178.                 <useUniqueVersions>false</useUniqueVersions>
  179.                 <classpathPrefix>lib/</classpathPrefix>
  180.                 <mainClass>cn.tedu.streaming.StreamingDriver</mainClass>
  181.               </manifest>
  182.             </archive>
  183.           </configuration>
  184.         </plugin>
  185.  
  186.  
  187.       </plugins>
  188.     </pluginManagement>
  189.   </build>
  190. </project>

7、学习Spark与Hbase整合基础

新建一个object

代码如下 

  1. package cn.yang.basic
  2.  
  3. import org.apache.hadoop.fs.shell.find.Result
  4. import org.apache.hadoop.hbase.client.Put
  5. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  6. import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
  7. import org.apache.hadoop.mapreduce.Job
  8. import org.apache.spark.{SparkConf, SparkContext}
  9.  
  10. /**
  11.  * 如何通过Spark将数据写出到HBase表中
  12.  */
  13. object HBaseWriteDriver {
  14.   def main(args: Array[String]): Unit = {
  15.     val conf = new SparkConf().setMaster("local").setAppName("writeHBase")
  16.     val sc = new SparkContext(conf)
  17.     //设定zookeeper集群IP地址。注意主机名和服务器ip对应一致
  18.     sc.hadoopConfiguration.set("hbase.zookeeper.quorum",
  19.     "hadoop01,hadoop02,hadoop03")
  20.     //设定zookeeper通信端口
  21.     sc.hadoopConfiguration.set("hbase.zookeeper.property.clientPort","2181")
  22.     //指定输出的HBase表名
  23.     sc.hadoopConfiguration.set(TableOutputFormat.OUTPUT_TABLE,"tbx")
  24.  
  25.     //创建Hadoop Job对象
  26.     val job = new Job(sc.hadoopConfiguration)
  27.     //设定输出的key类型
  28.     job.setOutputKeyClass(classOf[ImmutableBytesWritable])
  29.     //设定输出的value类型,导包:org.apache.hadoop.fs.shell.find.Result
  30.     job.setOutputValueClass(classOf[Result])
  31.     //设定输出表类型
  32.     job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])
  33.     val rdd=sc.parallelize(List("1 tom 23","2 rose 18","3 jim 25","4 jary 30"))
  34.  
  35.     //为了能够将数据插入到HBase表,需要做类型转换 RDD[String]->RDD[(输出key,输出value)]
  36.     val hbaseRDD=rdd.map{line=>
  37.       val arr = line.split(" ")
  38.       val id=arr(0)
  39.       val name=arr(1)
  40.       val age=arr(2)
  41.       //创建HBase的行对象并指定行键。导包:org.apache.hadoop.hbase.client.Put
  42.       val row = new Put(id.getBytes())
  43.       //①参:列族名  ②参:列名  ③参:列值
  44.       row.addColumn("cf1".getBytes(),"name".getBytes(),name.getBytes())
  45.       row.addColumn("cf1".getBytes(),"age".getBytes(),age.getBytes())
  46.  
  47.       (new ImmutableBytesWritable(),row)
  48.     }
  49.     //执行插入
  50.     hbaseRDD.saveAsNewAPIHadoopDataset(job.getConfiguration)
  51.   }
  52. }

8、启动服务器，三台zookeeper，hadoop，hbase

cd /home/software/hbase-2.4.2/bin/
sh start-hbase.sh 

9、进入01的hbase的shell建表 sh hbase shell

10、执行代码，附上读取与扫描hbase代码：

  1. package cn.yang.basic
  2.  
  3. import org.apache.hadoop.hbase.HBaseConfiguration
  4. import org.apache.hadoop.hbase.client.Result
  5. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  6. import org.apache.hadoop.hbase.mapreduce.TableInputFormat
  7. import org.apache.spark.{SparkConf, SparkContext}
  8.  
  9. object HBaseReadDriver {
  10.  
  11.   def main(args: Array[String]): Unit = {
  12.  
  13.     val conf=new SparkConf().setMaster("local").setAppName("read")
  14.  
  15.     val sc=new SparkContext(conf)
  16.  
  17.     //创建HBase环境参数对象
  18.     val hbaseConf=HBaseConfiguration.create()
  19.  
  20.     hbaseConf.set("hbase.zookeeper.quorum","hadoop01,hadoop02,hadoop03")
  21.     hbaseConf.set("hbase.zookeeper.property.clientPort","2181")
  22.     //指定读取的表名
  23.     hbaseConf.set(TableInputFormat.INPUT_TABLE,"tbx")
  24.  
  25.     //执行读取。并将HBase表数据读取到RDD结果集中
  26.     val resultRDD=sc.newAPIHadoopRDD(hbaseConf,
  27.       classOf[TableInputFormat],
  28.       classOf[ImmutableBytesWritable],
  29.       //导包:import org.apache.hadoop.hbase.client.Result
  30.       classOf[Result])
  31.  
  32.     resultRDD.foreach{case(k,v)=>
  33.       val name=v.getValue("cf1".getBytes(),"name".getBytes())
  34.       val age=v.getValue("cf1".getBytes(),"age".getBytes())
  35.  
  36.       println(new String(name)+":"+new String(age))
  37.     }
  38.   }
  39. }
  1. package cn.yang.basic
  2.  
  3. import org.apache.commons.codec.binary.Base64
  4. import org.apache.hadoop.hbase.HBaseConfiguration
  5. import org.apache.hadoop.hbase.client.{Result, Scan}
  6. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  7. import org.apache.hadoop.hbase.mapreduce.TableInputFormat
  8. import org.apache.hadoop.hbase.protobuf.ProtobufUtil
  9. import org.apache.spark.{SparkConf, SparkContext}
  10.  
  11. /**
  12.  * 学习如何扫描hbase表数据
  13.  */
  14. object HBaseScanDriver {
  15.  
  16.   def main(args: Array[String]): Unit = {
  17.  
  18.     val conf=new SparkConf().setMaster("local").setAppName("read")
  19.  
  20.     val sc=new SparkContext(conf)
  21.  
  22.     //创建HBase环境参数对象
  23.     val hbaseConf=HBaseConfiguration.create()
  24.  
  25.     hbaseConf.set("hbase.zookeeper.quorum","hadoop01,hadoop02,hadoop03")
  26.     hbaseConf.set("hbase.zookeeper.property.clientPort","2181")
  27.     //指定读取的表名
  28.     hbaseConf.set(TableInputFormat.INPUT_TABLE,"tbx")
  29.  
  30.     //创建HBase scan扫描对象
  31.     val scan=new Scan()
  32.     //设定扫描的起始行键
  33.     scan.setStartRow("2".getBytes())
  34.     //设定扫描终止行键。含头不含尾
  35.     scan.setStopRow("4".getBytes())
  36.  
  37.     //设定scan对象使其生效
  38.     hbaseConf.set(TableInputFormat.SCAN,
  39.       Base64.encodeBase64String(ProtobufUtil.toScan(scan).toByteArray()))
  40.  
  41.  
  42.  
  43.     //执行读取。并将HBase表数据读取到RDD结果集中
  44.     val resultRDD=sc.newAPIHadoopRDD(hbaseConf,
  45.       classOf[TableInputFormat],
  46.       classOf[ImmutableBytesWritable],
  47.       //导包:import org.apache.hadoop.hbase.client.Result
  48.       classOf[Result])
  49.  
  50.     resultRDD.foreach{case(k,v)=>
  51.       val name=v.getValue("cf1".getBytes(),"name".getBytes())
  52.       val age=v.getValue("cf1".getBytes(),"age".getBytes())
  53.  
  54.       println(new String(name)+":"+new String(age))
  55.     }
  56.   }
  57. }

2、实时流业务处理

实现步骤：

①、启动三台服务器，启动zookeeper，启动hadoop，启动kafka，启动flume

cd /home/software/kafka_2.10-0.10.0.1/bin/
sh kafka-server-start.sh ../config/server.properties

在flume的data目录，执行下面语句启动flume：

../bin/flume-ng agent -n a1 -c ./ -f ./weblog.conf -Dflume.root.logger=INFO,console

②、整合SparkStreaming与kafka，完成代码编写

在FluxStreamingServer下的scala文件下新建一个包，streaming，新建一个Driver

添加代码，下面为全部内容的代码：

结构：

Driver： 

  1. package cn.yang.streaming
  2.  
  3. import cn.yang.TongjiBean
  4. import cn.yang.dao.{HBaseUtil, MysqlUtil}
  5. import cn.yang.pojo.LogBean
  6. import org.apache.kafka.common.serialization.StringDeserializer
  7. import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
  8. import org.apache.spark.streaming.kafka010.KafkaUtils
  9. import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
  10. import org.apache.spark.streaming.{Seconds, StreamingContext}
  11. import org.apache.spark.{SparkConf, SparkContext}
  12.  
  13. import java.util.Calendar
  14.  
  15. object Driver {
  16.   def main(args: Array[String]): Unit = {
  17.     //如果后续要使用SparkStreaming从kafka消费数据，启动的线程数至少是2个
  18.     //其中一个线程负责SparkStreaming，另外一个线程负责从kafka消费数据
  19.     //还需要设定一下序列化参数
  20.     val conf=new SparkConf().setMaster("local[2]").setAppName("kafkasource")
  21.       .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
  22.  
  23.     val sc= new SparkContext(conf)
  24.     sc.setLogLevel("error")
  25.     //创建SparkStreaming对象，并指定批大小
  26.     val ssc = new StreamingContext(sc,Seconds(5))
  27.  
  28.     //指定从kafka消费的主题，通过Array可以指定消费多个主题
  29.     val topics = Array("fluxdata")
  30.  
  31.     //指定kafka的配置参数。通过Map来进行设定，key是属性名，value是属性值
  32.     //需要指定:kafka服务集群列表；key value的序列化类型，固定为String类型，消费者组名
  33.     val kafkaParams: Map[String, Object] = Map[String, Object](
  34.       "bootstrap.servers" -> "hadoop01:9092,hadoop02:9092,hadoop03:9092",
  35.       "key.deserializer" -> classOf[StringDeserializer],
  36.       "value.deserializer" -> classOf[StringDeserializer],
  37.       "group.id" -> "bos"
  38.     )
  39.  
  40.     //1参：SparkStreaming对象 2参：从Kafka消费模式，消费指定主题的所有分区数据
  41.     //3参：kafka订阅参阅信息
  42.     val stream=KafkaUtils.createDirectStream[String, String](
  43.       ssc,
  44.       PreferConsistent,
  45.       Subscribe[String, String](topics, kafkaParams))
  46.       .map(x=>x.value())
  47.  
  48.     //打印输出方式一
  49.     //foreachRDD,将当前批次内的所有数据转变为一个RDD
  50.     stream.foreachRDD{rdd=>
  51.       //将RDD[String]->Iterator[String]迭代器
  52.       val lines=rdd.toLocalIterator
  53.       //遍历迭代器
  54.       while(lines.hasNext){
  55.         //获取一条数据
  56.         val line=lines.next()
  57.         //第一步：做数据字段清洗。所需字段：url urlname uvid ssid sscount sstime cip
  58.         val arr= line.split("\\|")
  59.         val url=arr(0)
  60.         val urlname=arr(1)
  61.         val uvid=arr(13)
  62.         val ssid=arr(14).split("_")(0)
  63.         val sscount=arr(14).split("_")(1)
  64.         val sstime=arr(14).split("_")(2)
  65.         val cip=arr(15)
  66.  
  67.         //第二步：将清洗好的字段封装到bean中
  68.         val logBean=LogBean(url,urlname,uvid,ssid,sscount,sstime,cip)
  69.  
  70.         //第三步：统计实时业务指标。有pv uv vv newip newcust
  71.         //这5个指标的统计结果定为两种情况：1或0
  72.         //3-1 pv:页面访问量。用户访问1次，就记作1个pv
  73.         val pv =1
  74.  
  75.         //3-2 uv:独立用户数。uv=1或uv=0，处理逻辑：
  76.         //①拿着当前记录的uvid去HBase表（webtable）查询当天的所有数据
  77.         //②、如果没查到此uvid的记录，则记uv=1
  78.         //③、如果查到了此uvid的记录，则记uv=0
  79.         //实现难点：
  80.         //如何查询Hbase表‘当天’的数据？
  81.         //查询的起始时间戳startTime=当天的凌晨零点时间戳
  82.         //查询的终止时间戳endTime=sstime
  83.  
  84.         val endTime=sstime.toLong
  85.         val calendar=Calendar.getInstance()
  86.         calendar.setTimeInMillis(endTime)
  87.         calendar.set(Calendar.HOUR,0)
  88.         calendar.set(Calendar.MINUTE,0)
  89.         calendar.set(Calendar.SECOND,0)
  90.         calendar.set(Calendar.MILLISECOND,0)
  91.         //获取当天凌晨零点的时间戳
  92.         val startTime=calendar.getTimeInMillis
  93.  
  94.         //如何判断当前记录中的uvid在HBase表是否出现过？
  95.         //可以使用HBase的行键过滤器来实现(使用HBase的行键正则过滤器)
  96.         val uvRegex="^\\d+_"+uvid+".*$"
  97.         val uvRDD=HBaseUtil.queryHBase(sc,startTime,endTime,uvRegex)
  98.         val uv=if(uvRDD.count()==0) 1 else 0
  99.         //3-3 vv:独立会话数。vv=1 或 vv=0 判断逻辑同uv
  100.         //只不过判断指标变为当前记录的ssid
  101.         val vvRegex="^\\d+_\\d+_"+ssid+".*$"
  102.         val vvResult=HBaseUtil.queryHBase(sc,startTime,endTime,vvRegex)
  103.         val vv=if(vvResult.count()==0) 1 else 0
  104.         //3-4：newip：新增ip。newip=1 或newip=0 判断逻辑：
  105.         //用当前记录中的ip去HBase表查询历史数据（包含当天）
  106.         //如果没查到，则newip=1.反之newip=0
  107.         val ipRegex="^\\d+_\\d+_\\d+_"+cip+".*$"
  108.         val ipResult=HBaseUtil.queryHBase(sc,startTime=0,endTime,ipRegex)
  109.         val newip=if(ipResult.count()==0) 1 else 0
  110.         //3-5 newcust:新增用户数。处理逻辑和newip相同
  111.         //判断指标更换为uvid。正则使用uvRegex
  112.         val custResult=HBaseUtil.queryHBase(sc,startTime=0,endTime,uvRegex)
  113.         val newcust=if(custResult.count()==0)1 else 0
  114.  
  115.         //第四步：将统计好的业务指标封装到bean中，然后插入到mysql数据库中
  116.         val tongjiBean=TongjiBean(sstime,pv,uv,vv,newip,newcust)
  117.         MysqlUtil.saveToMysql(tongjiBean)
  118.  
  119.    //将封装好的bean数据存到HBase表中，供后续做查询使用
  120.         HBaseUtil.saveToHBase(sc,logBean)
  121.         println(logBean)
  122.       }
  123.     }
  124.     //打印输出方式二，有线-----
  125.     //stream.print()
  126.     ssc.start()
  127.     //保证SparkStreaming一直开启，直到用户主动中断退出为止
  128.     ssc.awaitTermination()
  129.  
  130.   }
  131. }

 dao-HBaseUtil

  1. package cn.yang.dao
  2.  
  3. import cn.yang.pojo.LogBean
  4. import org.apache.commons.codec.binary.Base64
  5. import org.apache.hadoop.fs.shell.find.Result
  6. import org.apache.hadoop.hbase.HBaseConfiguration
  7. import org.apache.hadoop.hbase.client.{Put, Scan}
  8. import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp
  9. import org.apache.hadoop.hbase.filter.{RegexStringComparator, RowFilter}
  10. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  11. import org.apache.hadoop.hbase.mapreduce.{TableInputFormat, TableOutputFormat}
  12. import org.apache.hadoop.hbase.protobuf.ProtobufUtil
  13. import org.apache.hadoop.mapreduce.Job
  14. import org.apache.spark.SparkContext
  15.  
  16. import scala.util.Random
  17.  
  18. object HBaseUtil {
  19.   def queryHBase(sc: SparkContext, startTime: Long, endTime: Long, regex: String) = {
  20.  
  21.     val hbaseConf=HBaseConfiguration.create()
  22.     hbaseConf.set("hbase.zookeeper.quorum", "hadoop01,hadoop02,hadoop03")
  23.     hbaseConf.set("hbase.zookeeper.property.clientPort","2181")
  24.     //指定读取的表名
  25.     hbaseConf.set(TableInputFormat.INPUT_TABLE,"webtable")
  26.  
  27.     val scan=new Scan()
  28.  
  29.     scan.withStartRow(startTime.toString().getBytes)
  30.     scan.withStopRow(endTime.toString().getBytes)
  31.  
  32.     //org.apache.hadoop.hbase.filter.RowFilter
  33.     val filter=new RowFilter(CompareOp.EQUAL,
  34.       new RegexStringComparator(regex))
  35.  
  36.     //绑定过滤器使其生效,即在做范围查询时结合行键正则过滤器来返回对应的结果
  37.     scan.setFilter(filter)
  38.  
  39.     //设置scan对象,使其生效
  40.     hbaseConf.set(TableInputFormat.SCAN,
  41.       Base64.encodeBase64String(ProtobufUtil.toScan(scan).toByteArray()))
  42.  
  43.     //执行读取,将结果返回到结果集RDD中
  44.     val resultRDD=sc.newAPIHadoopRDD(hbaseConf,
  45.       classOf[TableInputFormat],
  46.       classOf[ImmutableBytesWritable],
  47.       classOf[org.apache.hadoop.hbase.client.Result])
  48.     //QueryByRangeAndRegex方法:返回结果集RDD
  49.     resultRDD
  50.   }
  51.  
  52.   /*
  53.   将封装好的logbean数据存到指定的HBase中
  54.    */
  55.   def saveToHBase(sc: SparkContext, logBean: LogBean) = {
  56.     sc.hadoopConfiguration.set("hbase.zookeeper.quorum",
  57.                               "hadoop01,hadoop02,hadoop03")
  58.     sc.hadoopConfiguration.set("hbase.zookeeper.property.clientPort",
  59.                                "2181")
  60.     sc.hadoopConfiguration.set(TableOutputFormat.OUTPUT_TABLE,"webtable")
  61.     val job = new Job(sc.hadoopConfiguration)
  62.     job.setOutputKeyClass(classOf[ImmutableBytesWritable])
  63.     job.setOutputValueClass(classOf[Result])
  64.     job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])
  65.     val rdd=sc.parallelize(List(logBean))
  66.     val hbaseRDD=rdd.map{bean=>
  67.       //本项目的行键设计为：sstime_uvid_ssid_cip_随机数字
  68.       //行键以时间戳开头，作用：让HBase按时间戳做升序排序，便于后续按时间段范围查询
  69.       //行键中还包含uvid，ssid，cip信息。便于统计处相关业务指标，比如uv，vv等
  70.       //随机数字满足散列原则
  71.       val rowKey=bean.sstime+"_"+bean.uvid+"_"+bean.ssid+"_"+bean.cip+"_"+Random.nextInt(100)
  72.       //创建一个HBase行对象并指定行键
  73.       val row=new Put(rowKey.getBytes)
  74.       row.addColumn("cf1".getBytes,"url".getBytes,bean.url.getBytes)
  75.       row.addColumn("cf1".getBytes,"urlname".getBytes,bean.urlname.getBytes)
  76.       row.addColumn("cf1".getBytes,"uvid".getBytes,bean.uvid.getBytes)
  77.       row.addColumn("cf1".getBytes,"ssid".getBytes,bean.ssid.getBytes)
  78.       row.addColumn("cf1".getBytes,"sscount".getBytes,bean.sscount.getBytes)
  79.       row.addColumn("cf1".getBytes,"sstime".getBytes,bean.sstime.getBytes)
  80.       row.addColumn("cf1".getBytes,"cip".getBytes,bean.cip.getBytes)
  81.       (new ImmutableBytesWritable,row)
  82.  
  83.     }
  84.     //执行写出
  85.     hbaseRDD.saveAsNewAPIHadoopDataset(job.getConfiguration)
  86.   }
  87.  
  88. }

dao-MysqlUtil

  1. package cn.yang.dao
  2.  
  3. import cn.yang.TongjiBean
  4. import com.mchange.v2.c3p0.ComboPooledDataSource
  5.  
  6. import java.sql.{Connection, PreparedStatement, ResultSet}
  7. import java.text.SimpleDateFormat
  8.  
  9. object MysqlUtil {
  10.   val c3p0=new ComboPooledDataSource()
  11.   def saveToMysql(tongjiBean: TongjiBean) = {
  12.     var conn:Connection=null
  13.     var ps1:PreparedStatement=null
  14.     var rs:ResultSet=null
  15.     var ps2:PreparedStatement=null
  16.     var ps3:PreparedStatement=null
  17.  
  18.     try{
  19.       conn=c3p0.getConnection
  20.       /*
  21.       处理逻辑：
  22.         1、查询mysql的tongji2表当天的数据
  23.         2、如果当天还没有数据，则做新增插入
  24.           如果当天已有数据，则作更新累加
  25.        */
  26.       //解析出当天的日期。格式如:2022-04-24
  27.       val sdf=new SimpleDateFormat("YYYY-MM-dd")
  28.       val todayTime=sdf.format(tongjiBean.sstime.toLong)
  29.       //先查询tongji2表。如果当天已经有数据了,则更新累加
  30.       //如果当天还没有数据,则新增插入
  31.       ps1=conn.prepareStatement("select * from tongji2 where reporttime=?")
  32.       ps1.setString(1,todayTime)
  33.  
  34.       //执行查询并返回结果集
  35.       rs=ps1.executeQuery()
  36.       if(rs.next()){
  37.         //当天已经有数据，则做更新累积
  38.         ps2=conn.prepareStatement(
  39.           "update tongji2 set pv=pv+?,uv=uv+?,vv=vv+?,newip=newip+?,newcust=newcust+? where reporttime=?")
  40.  
  41.         ps2.setInt(1, tongjiBean.pv)
  42.         ps2.setInt(2, tongjiBean.uv)
  43.         ps2.setInt(3, tongjiBean.vv)
  44.         ps2.setInt(4, tongjiBean.newip)
  45.         ps2.setInt(5, tongjiBean.newcust)
  46.         ps2.setString(6, todayTime)
  47.  
  48.         ps2.executeUpdate()
  49.       }else{
  50.         //则表示当天还没有数据,则新增插入
  51.         ps3=conn.prepareStatement("insert into tongji2 values(?,?,?,?,?,?)")
  52.         ps3.setString(1, todayTime)
  53.         ps3.setInt(2, tongjiBean.pv)
  54.         ps3.setInt(3, tongjiBean.uv)
  55.         ps3.setInt(4, tongjiBean.vv)
  56.         ps3.setInt(5, tongjiBean.newip)
  57.         ps3.setInt(6, tongjiBean.newcust)
  58.  
  59.         ps3.executeUpdate()
  60.       }
  61.     }catch {
  62.       case t:Exception=>{
  63.         t.printStackTrace()
  64.       }
  65.     }finally {
  66.       if(ps3!=null)ps3.close
  67.       if(ps2!=null)ps2.close
  68.       if(rs!=null)rs.close
  69.       if(ps1!=null)ps1.close
  70.       if(conn!=null)conn.close
  71.     }
  72.   }
  73.  
  74. }

pojo-LogBean

  1. package cn.yang.pojo
  2.  
  3. case class LogBean(url:String,
  4.                    urlname:String,
  5.                    uvid:String,
  6.                    ssid:String,
  7.                    sscount:String,
  8.                    sstime:String,
  9.                    cip:String)

TongjiBean

  1. package cn.yang
  2.  
  3. case class TongjiBean(sstime:String,
  4.                       pv:Int,
  5.                       uv:Int,
  6.                       vv:Int,
  7.                       newip:Int,
  8.                       newcust:Int)

③、启动SparkStreaming

④、启动tomcat，访问埋点服务器，测试SparkStreaming是否能够收到数据

⑤、启动HBase

cd /home/software/hbase-2.4.2/bin/

sh start-hbase.sh

sh hbase shell

建表：create 'webtable','cf1'

 ⑥、我们启动tomcat，启动Driver测试，然后扫描表webtable

发现有数据了：

⑦、进入mysql，在weblog库下新建表

 create table tongji2(reporttime date,pv int,uv int,vv int,newip int,newcust int);

⑧、执行我们的程序，访问埋点，最后到mysql查看数据，这也是我们本项目实现的最终结果，结果内容存储到mysql数据库中

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

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

  

闽ICP备14008679号