1 项目需求
1）需求
- 统计今天到目前为止的访问量
- 统计今天到目前为止从搜索引擎过来的课程的访问量
2）开发环境与技术选型
- IDEA+maven
- flume+kafka+HBase
3）安装配置 HBase
- 下载、解压、配置环境变量
- 配置文件
conf/hbase-env.sh
修改JAVA_HOMEexport HBASE_MANAGES_ZK=falseconf/hbase-site.xml
<configuration><property><name>hbase.rootdir</name><value>hdfs://localhost:8020/hbase</value></property><property><name>hbase.cluster.distributed</name><value>true</value></property><property><name>hbase.zookeeper.quorum</name><value>localhost:2181</value></property></configuration>conf/regionservers
localhost4）HBase 建表
//1 启动hbase
start-hbase.sh
// 2 启动shell
hbaseshell
// 3 建表create'course_clickcount','info'create'course_search_clickcount','info'
// 4 查看数据表
list
// 5 查看数据表信息
describe'course_clickcount'
// 6 查看表数据
scan'course_clickcount'5）代码地址
2 模拟日志生成
1）使用python开发日志生成器模拟产生日志，每分钟产生一次日志信息
generate_log.py
#coding=UTF-8import randomimport time
url_paths=["class/112.html","class/128.html","class/145.html","class/130.html","class/146.html","class/131.html","learn/821","course/list"
]
ip_slices=[132,156,124,10,29,167,143,187,30,46,55,63,72,87,98,168]
http_referers=["https://www.baidu.com/s?wd={query}","https://www.sogou.com/web?query={query}","https://cn.bing.com/search?q={query}","https://www.so.com/s?q={query}"
]
search_keyword=["spark sql实战","hadoop 基础","storm实战","spark streaming实战"
]
status_code=["200","404","500"]defsample_status_code():return random.sample(status_code,1)[0]defsample_referer():if random.uniform(0,1)>0.2:return"-"
refer_str=random.sample(http_referers,1)
query_str=random.sample(search_keyword,1)return refer_str[0].format(query=query_str[0])defsample_url():return random.sample(url_paths,1)[0]defsample_ip():
slice=random.sample(ip_slices,4)return".".join([str(item)for itemin slice])defgenerate_log(count=10):
time_str=time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())
f=open("/Users/Mac/testdata/streaming_access.log","w+")while count >=1:
query_log="{ip}\t{local_time}\t\"GET /{url} HTTP/1.1\"\t{status_code}\t{refer}".format(url=sample_url(),ip=sample_ip(),refer=sample_referer(),status_code=sample_status_code(),local_time=time_str)
print(query_log)
f.write(query_log+"\n")
count=count-1if __name__ =='__main__':# 每一分钟生成一次日志信息whileTrue:
generate_log()
time.sleep(60)3 flume收集日志并对接kafka
1）编写flume配置文件，streaming_project2.conf
exec-memory-kafka.sources= exec-source
exec-memory-kafka.sinks=kafka-sink
exec-memory-kafka.channels= memory-channel
exec-memory-kafka.sources.exec-source.type=exec
exec-memory-kafka.sources.exec-source.command=tail-F /Users/Mac/testdata/streaming_access.log
exec-memory-kafka.sources.exec-source.shell= /bin/sh-c
exec-memory-kafka.memory-channel.type=memory
exec-memory-kafka.sinks.kafka-sink.type= org.apache.flume.sink.kafka.KafkaSink
exec-memory-kafka.sinks.kafka-sink.brokerList=localhost:9092
exec-memory-kafka.sinks.kafka-sink.topic=test_topic
exec-memory-kafka.sinks.kafka-sink.batchSize=5
exec-memory-kafka.sinks.kafka-sink.requireedAcks=1
exec-memory-kafka.sources.exec-source.channels=memory-channel
exec-memory-kafka.sinks.kafka-sink.channel= memory-channel4 业务开发
4.1 消费kafka数据、数据清洗与统计
1）实体类
ClickLog.scala
package com.lihaogn.sparkProject.domain/**
* 清洗后的日志格式
*
*@param ip
*@param time
*@param courseId
*@param statusCode 日志访问状态码
*@param referer
*/caseclassClickLog(ip: String, time: String, courseId: Int, statusCode: Int, referer: String)CourseClickCount.scala
package com.lihaogn.sparkProject.domain/**
* 课程点击次数实体类
*
*@param day_course 对应HBase中的rowkey
*@param click_count 访问次数
*/caseclassCourseClickCount(day_course: String, click_count: Long)CourseSearchClickCount.scala
package com.lihaogn.sparkProject.domain/**
* 从搜索引擎过来的课程点击数实体类
*@param day_search_course
*@param click_count
*/caseclassCourseSearchClickCount(day_search_course: String, click_count: Long)2）工具类
DateUtils.scala
package com.lihaogn.sparkProject.utilsimport java.util.Dateimport org.apache.commons.lang3.time.FastDateFormat/**
* 日期时间工具类
*/objectDateUtils {val OLD_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")val TARGET_FORMAT = FastDateFormat.getInstance("yyyyMMddHHmmss")def getTime(time: String) = {
OLD_FORMAT.parse(time).getTime
}def parseToMinute(time: String) = {
TARGET_FORMAT.format(new Date(getTime(time)))
}def main(args: Array[String]): Unit = {
println(parseToMinute("2018-9-6 13:58:01"))
}
}添加依赖
<!-- cloudera repo--><repositories><repository><id>cloudera</id><url>https://repository.cloudera.com/artifactory/cloudera-repos</url></repository></repositories><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>${hbase.version}</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>${hbase.version}</version></dependency><!-- hadoop --><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency>HBaseUtils.java
package com.lihaogn.spark.project.utils;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.hbase.client.HBaseAdmin;import org.apache.hadoop.hbase.client.HTable;import org.apache.hadoop.hbase.client.Put;import org.apache.hadoop.hbase.util.Bytes;import java.io.IOException;/**
* HBase操作工具类，Java工具类建议采用单例模式封装
*/publicclassHBaseUtils {
HBaseAdmin admin =null;
Configuration configuration =null;/**
* 私有构造方法
*/privateHBaseUtils() {
configuration =new Configuration();
configuration.set("hbase.zookeeper.quorum","localhost:2181");
configuration.set("hbase.rootdir","hdfs://localhost:8020/hbase");try {
admin =new HBaseAdmin(configuration);
}catch (IOException e) {
e.printStackTrace();
}
}privatestatic HBaseUtils instance =null;publicstaticsynchronized HBaseUtilsgetInstance() {if (null == instance) {
instance =new HBaseUtils();
}return instance;
}/**
* 根据表名获取到HTable实例
*
* @param tableName
* @return
*/public HTablegetTable(String tableName) {
HTable table =null;try {
table =new HTable(configuration, tableName);
}catch (IOException e) {
e.printStackTrace();
}return table;
}/**
* 添加一条记录到表中
*
* @param tableName
* @param rowkey
* @param cf
* @param column
* @param value
*/publicvoidput(String tableName, String rowkey, String cf, String column, String value) {
HTable table = getTable(tableName);
Put put =new Put(Bytes.toBytes(rowkey));
put.add(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(value));try {
table.put(put);
}catch (IOException e) {
e.printStackTrace();
}
}publicstaticvoidmain(String[] args) {// HTable table = HBaseUtils.getInstance().getTable("course_clickcount");// System.out.println(table.getName().getNameAsString());
String tableName ="course_clickcount";
String rowkey ="20180906_1";
String cf ="info";
String column ="click_count";
String value ="2";
HBaseUtils.getInstance().put(tableName, rowkey, cf, column, value);
}
}3）数据库操作
CourseClickCountDAO.scala
package com.lihaogn.sparkProject.daoimport com.lihaogn.spark.project.utils.HBaseUtilsimport com.lihaogn.sparkProject.domain.CourseClickCountimport org.apache.hadoop.hbase.client.Getimport org.apache.hadoop.hbase.util.Bytesimport scala.collection.mutable.ListBuffer/**
* 数据访问层，课程点击数
*/objectCourseClickCountDAO {val tableName ="course_clickcount"val cf ="info"val qualifer ="click_count"/**
* 保存数据到HBase
*
*@param list
*/def save(list: ListBuffer[CourseClickCount]): Unit = {val table = HBaseUtils.getInstance().getTable(tableName)for (ele <- list) {
table.incrementColumnValue(Bytes.toBytes(ele.day_course),
Bytes.toBytes(cf),
Bytes.toBytes(qualifer),
ele.click_count)
}
}/**
* 根据rowkey查询值
*@param day_course
*@return
*/def count(day_course:String):Long= {val table = HBaseUtils.getInstance().getTable(tableName)val get =new Get(Bytes.toBytes(day_course))val value = table.get(get).getValue(cf.getBytes, qualifer.getBytes)if (value ==null) {0L
}else
Bytes.toLong(value)
}def main(args: Array[String]): Unit = {val list=new ListBuffer[CourseClickCount]
list.append(CourseClickCount("20180906_8",8))
list.append(CourseClickCount("20180906_4",3))
list.append(CourseClickCount("20180906_2",2))
save(list)
println(count("20180906_8")+":"+count("20180906_4")+":"+count("20180906_2"))
}
}CourseSearchClickCountDAO.scala
package com.lihaogn.sparkProject.daoimport com.lihaogn.spark.project.utils.HBaseUtilsimport com.lihaogn.sparkProject.domain.{CourseClickCount, CourseSearchClickCount}import org.apache.hadoop.hbase.client.Getimport org.apache.hadoop.hbase.util.Bytesimport scala.collection.mutable.ListBuffer/**
* 数据访问层，从搜索引擎过来的课程点击数
*/objectCourseSearchClickCountDAO {val tableName ="course_search_clickcount"val cf ="info"val qualifer ="click_count"/**
* 保存数据到HBase
*
*@param list
*/def save(list: ListBuffer[CourseSearchClickCount]): Unit = {val table = HBaseUtils.getInstance().getTable(tableName)for (ele <- list) {
table.incrementColumnValue(Bytes.toBytes(ele.day_search_course),
Bytes.toBytes(cf),
Bytes.toBytes(qualifer),
ele.click_count)
}
}/**
* 根据rowkey查询值
*
*@param day_search_course
*@return
*/def count(day_search_course: String): Long = {val table = HBaseUtils.getInstance().getTable(tableName)val get =new Get(Bytes.toBytes(day_search_course))val value = table.get(get).getValue(cf.getBytes, qualifer.getBytes)if (value ==null) {0L
}else
Bytes.toLong(value)
}def main(args: Array[String]): Unit = {val list =new ListBuffer[CourseSearchClickCount]
list.append(CourseSearchClickCount("20180906_www.baidu.com_8",8))
list.append(CourseSearchClickCount("20180906_www.baidu.com_4",3))
save(list)
println(count("20180906_www.baidu.com_8") +":" + count("20180906_www.baidu.com_4"))
}
}4）主类
SparkStreamingApp.scala
packagecom.lihaogn.sparkProject.main
importcom.lihaogn.sparkProject.dao.{CourseClickCountDAO, CourseSearchClickCountDAO}
importcom.lihaogn.sparkProject.domain.{ClickLog, CourseClickCount, CourseSearchClickCount}
importcom.lihaogn.sparkProject.utils.DateUtils
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka.KafkaUtils
import scala.collection.mutable.ListBuffer/**
* 使用spark streaming分析日志
*/
object SparkStreamingApp {
def main(args: Array[String]): Unit = {
if (args.length !=4) {
System.err.println("usage: KafKaReceiverWC <zkQuorum> <group> <topics> <numThreads>")
}
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf().setAppName("SparkStreamingApp").setMaster("local[5]")
val ssc = new StreamingContext(sparkConf, Seconds(5))
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
// spark streaming 对接 kafka
val messages = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)
// 步骤一：测试数据接收
messages.map(_._2).count().print()
// 步骤二：数据清洗
val logs = messages.map(_._2)
val cleanData = logs.map(line => {
val infos = line.split("\t")
val url = infos(2).split(" ")(1)
var courseId =0
// 获取课程标号
if (url.startsWith("/class")) {
val courseHtml = url.split("/")(2)
courseId = courseHtml.substring(0, courseHtml.lastIndexOf(".")).toInt
}
ClickLog(infos(0), DateUtils.parseToMinute(infos(1)), courseId, infos(3).toInt, infos(4))
}).filter(clicklog => clicklog.courseId !=0)
cleanData.print()
// 步骤三：统计今天到现在为止的课程访问量
cleanData.map(x=>{
(x.time.substring(0,8)+"_"+x.courseId,1)
}).reduceByKey(_+_).foreachRDD(rdd=>{
rdd.foreachPartition(partitionRecords=>{
val list=new ListBuffer[CourseClickCount]
partitionRecords.foreach(pair=>{
list.append(CourseClickCount(pair._1,pair._2))
})
// 写入数据库
CourseClickCountDAO.save(list)
})
})
// 步骤四：统计从搜索引擎过来的从今天开始到现在的课程的访问量
cleanData.map(x=>{
val referer=x.referer.replaceAll("//","/")
val splits=referer.split("/")
var host=""
if(splits.length>2) {
host=splits(1)
}
(host,x.courseId,x.time)
}).filter(_._1!="").map(x=>{
(x._3.substring(0,8)+"_"+x._1+"_"+x._2,1)
}).reduceByKey(_+_).foreachRDD(rdd=>{
rdd.foreachPartition(partitionRecords=>{
val list =new ListBuffer[CourseSearchClickCount]
partitionRecords.foreach(pair=>{
list.append(CourseSearchClickCount(pair._1,pair._2))
})
// 写入数据库
CourseSearchClickCountDAO.save(list)
})
})
ssc.start()
ssc.awaitTermination()
}
}5 运行测试
1）启动 zookeeper
zkServer.shstart2）启动 HDFS
start-dfs.shstart-yarn.sh3）启动 kafka
kafka-server-start.sh-daemon$KAFKA_HOME/config/server.properties $4）启动 flume
flume-ng agent \
--conf$FLUME_HOME/conf \--conf-file$FLUME_HOME/conf/streaming_project2.conf \--name exec-memory-kafka \-Dflume.root.logger=INFO,console5）运行日志生成器
python3 generate_log.py6）运行spark程序
spark-submit \
--classcom.lihaogn.sparkProject.main.SparkStreamingApp \--master local[5] \--nameSparkStreamingApp \--jars /Users/Mac/software/spark-streaming-kafka-0-8-assembly_2.11-2.2.0.jar,$(echo /Users/Mac/app/hbase-1.2.0-cdh5.7.0/lib/*.jar | tr' '',') \/Users/Mac/my-lib/Kafka-train-1.0.jar \
localhost:2181 test test_topic 17）结果
