前言
本人GitHub地址：https://github.com/guofei1219
QQ : 86608625
咨詢項(xiàng)目相關(guān)問題的請(qǐng)直接說明問題，不要一直問在嗎叮贩？還在嗎几于？等問題，博主QQ一直健在呢筹陵，由于本人平時(shí)還要工作诲宇，問題不能及時(shí)回復(fù)請(qǐng)見諒！；谭姑蓝！
背景
用戶下單數(shù)據(jù)會(huì)通過業(yè)務(wù)系統(tǒng)實(shí)時(shí)產(chǎn)生入庫(kù)到mysql庫(kù)，我們要統(tǒng)計(jì)通某個(gè)推廣渠道實(shí)時(shí)下單量吕粗，以便線上運(yùn)營(yíng)推廣人員查看不同渠道推廣效果進(jìn)而執(zhí)行不同推廣策略
系統(tǒng)架構(gòu)
注：組件不了解的同學(xué)可參考其他文章纺荧，本文主要講項(xiàng)目的實(shí)現(xiàn)
1、某些同學(xué)會(huì)問颅筋，直接在業(yè)務(wù)系統(tǒng)加入JS埋點(diǎn)通過發(fā)日志不更好嗎宙暇？
答：第一、JS埋點(diǎn)業(yè)務(wù)系統(tǒng)涉及產(chǎn)品改造议泵，不可能因?yàn)橐粋€(gè)需求讓你去隨便改業(yè)務(wù)系統(tǒng)占贫。第二、即使加入JS埋點(diǎn)也不可能獲得業(yè)務(wù)系統(tǒng)的全部數(shù)據(jù)先口。所以業(yè)務(wù)系統(tǒng)核心數(shù)據(jù)還得去業(yè)務(wù)系統(tǒng)庫(kù)獲取型奥。
2、還有人問加入Kafka太多余
答：第一碉京、加入Kafka為了使系統(tǒng)擴(kuò)展性更強(qiáng)厢汹，可方便對(duì)接各種開源產(chǎn)品。第二谐宙、通過Kafka消息組可使同一條消息被不同Consumer消費(fèi)烫葬，用戶離線和實(shí)時(shí)兩條線。
解析Mysql binlog日志
主要邏輯
1.創(chuàng)建Canal連接
2.解析Mysql binlog獲得insert語句
public static void main(String args[]) {
//第一個(gè)參數(shù)為Canal server服務(wù)IP地址如果使用windows開發(fā)連接linux Canal服務(wù)需要制定IP eg: new InetSocketAddress("192.168.61.132", 11111)
//第二個(gè)參數(shù)為Canal server服務(wù)端口號(hào) Canal server IP和端口號(hào)在 /conf/canal.properties中配置
//第三個(gè)參數(shù)為Canal instance名稱 /conf下目錄名稱
//第四第五個(gè)參數(shù)為mysql用戶名和密碼，如果在 /conf/example/instance.properties中已經(jīng)配置 這里不用謝
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.61.132",
11111), "example", "", "");
int batchSize = 1000;
int emptyCount = 0;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
int totalEmtryCount = 120;
while (emptyCount < totalEmtryCount) {
Message message = connector.getWithoutAck(batchSize); // 獲取指定數(shù)量的數(shù)據(jù)
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
emptyCount++;
System.out.println("empty count : " + emptyCount);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
emptyCount = 0;
// System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
printEntry(message.getEntries());
}
connector.ack(batchId); // 提交確認(rèn)
// connector.rollback(batchId); // 處理失敗, 回滾數(shù)據(jù)
}
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
}
}
組裝數(shù)據(jù)發(fā)送至Kafka
private static void printColumn(List<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue());
KafkaProducer.sendMsg("canal", UUID.randomUUID().toString() ,column.getName() + " : " + column.getValue());
}
}
Streaming分渠道匯總數(shù)據(jù)
以DStream中的數(shù)據(jù)進(jìn)行按key做reduce操作搭综，然后對(duì)各個(gè)批次的數(shù)據(jù)進(jìn)行累加
在有新的數(shù)據(jù)信息進(jìn)入或更新時(shí)垢箕，可以讓用戶保持想要的任何狀。使用這個(gè)功能需要完成兩步：
- 定義狀態(tài)：可以是任意數(shù)據(jù)類型
- 定義狀態(tài)更新函數(shù)：用一個(gè)函數(shù)指定如何使用先前的狀態(tài)兑巾，從輸入流中的新值更新狀態(tài)舰讹。
對(duì)于有狀態(tài)操作，要不斷的把當(dāng)前和歷史的時(shí)間切片的RDD累加計(jì)算闪朱，隨著時(shí)間的流失月匣，計(jì)算的數(shù)據(jù)規(guī)模會(huì)變得越來越大。
val orders = resut_lines.updateStateByKey(updateRunningSum _)
def updateRunningSum(values: Seq[Long], state: Option[Long]) = {
/*
state:存放的歷史數(shù)據(jù)
values:當(dāng)前批次匯總值
*/
Some(state.getOrElse(0L)+values.sum)
}
統(tǒng)計(jì)結(jié)果寫入Mysql
實(shí)時(shí)匯總某渠道下單量需要根據(jù)渠道為主鍵更新或插入新數(shù)據(jù)
1.當(dāng)某個(gè)渠道第一單時(shí)奋姿，庫(kù)中沒有以此渠道為主鍵的數(shù)據(jù)锄开，需要insert into 訂單統(tǒng)計(jì)表
2.當(dāng)某渠道在庫(kù)中已有該渠道下單量，需要更新此渠道下單量值 update 訂單統(tǒng)計(jì)表
所以我們使用：
#有該渠道就更新称诗，沒有就插入
REPLACE INTO order_statistic(chanel, orders) VALUES(?, ?)
orders.foreachRDD(rdd =>{
rdd.foreachPartition(rdd_partition =>{
rdd_partition.foreach(data=>{
if(!data.toString.isEmpty) {
System.out.println("訂單量"+" : "+data._2)
DataUtil.toMySQL(data._1.toString,data._2.toInt)
}
})
})
})
def toMySQL(name: String,orders:Int) = {
var conn: Connection = null
var ps: PreparedStatement = null
val sql = "REPLACE INTO order_statistic(chanel, orders) VALUES(?, ?)"
try {
Class.forName("com.mysql.jdbc.Driver");
conn = DriverManager.getConnection("jdbc:mysql://192.168.20.126:3306/test", "root", "root")
ps = conn.prepareStatement(sql)
ps.setString(1, name)
ps.setInt(2, orders)
ps.executeUpdate()
} catch {
case e: Exception => e.printStackTrace()
} finally {
if (ps != null) {
ps.close()
}
if (conn != null) {
conn.close()
}
}
}
FAQ
1.canal依賴Canal protobuf版本為2.4.1萍悴，而spark依賴的2.5版本
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>2.4.1</version>
</dependency>
參考文章
1.Canal wiki:
https://github.com/alibaba/canal/wiki
2.streaming關(guān)于轉(zhuǎn)化操作
http://spark.apache.org/docs/1.6.0/streaming-programming-guide.html#transformations-on-dstreams
3.mysql的replace into
http://blog.sina.com.cn/s/blog_5f53615f01016wy3.html
