Spark實(shí)時(shí)統(tǒng)計(jì)訂單量 - 簡書
http://www.reibang.com/p/3ec093a9d584
注：組件不了解的同學(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)庫獲取鸳玩。
2、還有人問加入Kafka太多余
答：第一糟描、加入Kafka為了使系統(tǒng)擴(kuò)展性更強(qiáng)怀喉，可方便對(duì)接各種開源產(chǎn)品。第二船响、通過Kafka消息組可使同一條消息被不同Consumer消費(fèi)躬拢，用戶離線和實(shí)時(shí)兩條線。
前言
本人GitHub地址：https://github.com/guofei1219QQ : 86608625咨詢項(xiàng)目相關(guān)問題的請(qǐng)直接說明問題见间，不要一直問在嗎聊闯？還在嗎？等問題米诉，博主QQ一直健在呢菱蔬，由于本人平時(shí)還要工作，問題不能及時(shí)回復(fù)請(qǐng)見諒Ｊ仿隆Ｋ┟凇！
背景
用戶下單數(shù)據(jù)會(huì)通過業(yè)務(wù)系統(tǒng)實(shí)時(shí)產(chǎn)生入庫到mysql庫惊橱，我們要統(tǒng)計(jì)通某個(gè)推廣渠道實(shí)時(shí)下單量蚪腐，以便線上運(yùn)營推廣人員查看不同渠道推廣效果進(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)庫獲取涵但。
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è)功能需要完成兩步：1) 定義狀態(tài)：可以是任意數(shù)據(jù)類型2) 定義狀態(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í)，庫中沒有以此渠道為主鍵的數(shù)據(jù)画畅，需要insert into 訂單統(tǒng)計(jì)表2.當(dāng)某渠道在庫中已有該渠道下單量砸琅，需要更新此渠道下單量值 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/wiki2.streaming關(guān)于轉(zhuǎn)化操作http://spark.apache.org/docs/1.6.0/streaming-programming-guide.html#transformations-on-dstreams3.mysql的replace intohttp://blog.sina.com.cn/s/blog_5f53615f01016wy3.html
文／MichaelFly（簡書作者）原文鏈接：http://www.reibang.com/p/3ec093a9d584著作權(quán)歸作者所有症脂，轉(zhuǎn)載請(qǐng)聯(lián)系作者獲得授權(quán)，并標(biāo)注“簡書作者”淫僻。
