package com.anryg.bigdata.streaming.ods
import com.alibaba.fastjson.JSON
import com.anryg.bigdata.streaming.StreamingProcessHelper
import org.apache.spark.SparkConf
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
/**
* @DESC:
* @Auther: Anryg
* @Date: 2022/8/31 19:03
*/
object StreamingSource2HiveOds extends StreamingProcessHelper[Any]{
/**
* @DESC: 主函数，应用运行入口
* */
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("StreamingSource2HiveOds").setMaster("local[*]")
val spark = SparkSession.builder().config(conf)
.config("spark.sql.hive.hiveserver2.jdbc.url","jdbc:hive2://server1:2181,server2:2181,server3:2181/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2")
.config("spark.datasource.hive.warehouse.metastoreUri","thrift://serverXXX:9083")
.enableHiveSupport() //打开hive支持功能，可以与hive共享catalog
.getOrCreate()
clickProcess(spark)
}
/**
*@DESC: 将kafka数据源数据读取出来成为DataFrame
* */
def readKafka2DF(sparkSession: SparkSession): DataFrame ={
val config = Map(("kafka.bootstrap.servers", "xxxx:6667"),("subscribe","test"),
("failOnDataLoss","false"),("fetchOffset.numRetries","3"),("startingOffsets","earliest"))
getStreamingReader(sparkSession,"kafka",config).load()
}
/**
*@DESC: 将原始DF进行业务逻辑处理
* */
def handleData(sparkSession: SparkSession, rawDF:DataFrame): DataFrame ={
import sparkSession.implicits._
val targetDS = rawDF.selectExpr("CAST(value AS STRING)") //将kafka中的数据的value转为为string，原始为binary类型
.map(row => {
val line = row.getAs[String]("value") //获取row对象中的field，其实也只有一个field
val rawJson = JSON.parseObject(line) //原始string是一个json，对其进行解析
val message = rawJson.getString("message") //获取业务数据部分
val msgArray = message.split(",") //指定分隔符进行字段切分
msgArray
}).filter(_.length == 9).filter(array => array(2).length >= 8)//确保日期字段符合规范
.map(array =>(array(0),array(1),array(2),array(3), array(4),array(5),array(6),array(7),array(8),
array(2).substring(0,4),array(2).substring(4,6),array(2).substring(6,8)))
.toDF("client_ip","domain","time","target_ip","rcode","query_type","authority_record","add_msg","dns_ip","year","month","day") //给裸数据添加字段名
targetDS
}
/**
*@DESC: 将目标数据集进行写入hive的ODS
* */
def sinkData(targetDS:DataFrame): StreamingQuery ={
val config = Map(("checkpointLocation","hdfs://XXXX:8020/tmp/offset/test/StreamingSource2HiveOds"),
("format","append"))
getStreamingWriter(targetDS, OutputMode.Append(),"orc",config)
.partitionBy("year","month","day")
.toTable("ods.ods_kafka_internetlog")
}
/**
* @DESC: 将所有数据步骤串起来
* */
def clickProcess(sparkSession: SparkSession): Unit ={
val rawDF = readKafka2DF(sparkSession)
val targetDS = handleData(sparkSession, rawDF)
sinkData(targetDS).awaitTermination()
}
}
package com.anryg.bigdata.streaming.dwd
import com.anryg.bigdata.{IpSearch, RedisClientUtils}
import com.anryg.bigdata.streaming.StreamingProcessHelper
import org.apache.spark.SparkConf
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
* @DESC: 读取ods层数据，并加工写入到dwd
* @Auther: Anryg
* @Date: 2022/9/1 09:53
*/
object StreamingFromOds2Dwd extends StreamingProcessHelper[Any]{
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("StreamingFromOds2Dwd").setMaster("local")
val spark = SparkSession.builder().config(conf)
.config("spark.sql.hive.hiveserver2.jdbc.url","jdbc:hive2://server1:2181,server2:2181,server3:2181/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2")
.config("spark.datasource.hive.warehouse.metastoreUri","thrift://XXXX:9083")
.enableHiveSupport() //打开hive支持功能，可以与hive共享catalog
.getOrCreate()
clickProcess(spark,"ods.ods_kafka_internetlog","dwd.dwd_internetlog_detail")
}
/**
*@DESC: 流方式读取hive数据源
* */
def readHive2DF(sparkSession: SparkSession, sourceTable:String): DataFrame ={
sparkSession.readStream.table(sourceTable)
}
/**
*@DESC: 对ods数据进行字段补齐等处理
* */
def handleData(sparkSession: SparkSession, dataFrame: DataFrame, tableName:String): DataFrame ={
import sparkSession.implicits._
dataFrame.printSchema()
dataFrame.map(row => {
val clientIP = row.getAs[String]("client_ip")
val ipAndAddr = IpSearch.getAddrByIP(RedisClientUtils.getSingleRedisClient,clientIP).split("-")
val country = ipAndAddr(2)
val province = ipAndAddr(3)
val city = ipAndAddr(4)
val operator = ipAndAddr(5)
val domain = row.getAs[String]("domain").toLowerCase//将域名转成小写
val time = row.getAs[String]("time")
val targetIP = row.getAs[String]("target_ip")
val rcode = row.getAs[String]("rcode")
val queryType = row.getAs[String]("query_type")
val authRecord = row.getAs[String]("authority_record").toLowerCase
val addMsg = row.getAs[String]("add_msg")
val dnsIP = row.getAs[String]("dns_ip")
val year = row.getAs[String]("year")
val month = row.getAs[String]("month")
val day = row.getAs[String]("day")
(clientIP,country,province,city,operator,domain,time,targetIP,rcode,queryType,authRecord,addMsg,dnsIP,year,month,day)
}).toDF("client_ip","country","province","city","operator","domain","time","target_ip","rcode","query_type","authority_record","add_msg","dns_ip","year","month","day")
}
/**
*@DESC: 将处理好的数据sink到dwd表中
* */
def sinkData(targetDS:DataFrame, tableName:String): StreamingQuery ={
val config = Map(("checkpointLocation","hdfs://XXXX:8020/tmp/offset/test/StreamingFromOds2Dwd"),
("format","append"))
getStreamingWriter(targetDS,OutputMode.Append(),"orc",config)
.partitionBy("year","month","day")
.toTable(tableName)
}
/**
* @DESC: 将所有数据步骤串起来
* */
def clickProcess(sparkSession: SparkSession,sourceTable:String, sinkTable:String): Unit ={
val rawDF = readHive2DF(sparkSession, sourceTable)
val targetDS = handleData(sparkSession, rawDF, sourceTable)
sinkData(targetDS, sinkTable).awaitTermination()
}
}
|最后
这部分代码都是我这几天现写的，整个流程也都实现了，当然，过程中不免遇到很多坑，具体有哪些坑、如何避坑，以及该部分内容的讲解，周日晚八点会在B站直播给大家唠唠，有兴趣的小伙伴，欢迎来捧场。
最后，该项目的github地址: https://github.com/Anryg/internet_behavior_project
有任何问题，欢迎找我交流哈。
原文作者：安瑞哥是码农；如有侵权，请联系管理员删除。
上一篇：
程序员需要掌握哪些隐形技能
下一篇：
一份能打动面试官的简历如何书写？
请登录后发表评论
登录
全部评论
