背景
用户下单数据会通过业务系统实时产生入库到mysql库，我们要统计通某个推广渠道实时下单量，以便线上运营推广人员查看不同渠道推广效果进而执行不同推广策略
系统架构
架构图
注：组件不了解的同学可参考其他文章，本文主要讲项目的实现
1、某些同学会问，直接在业务系统加入JS埋点通过发日志不更好吗？
答：第一、JS埋点业务系统涉及产品改造，不可能因为一个需求让你去随便改业务系统。第二、即使加入JS埋点也不可能获得业务系统的全部数据。所以业务系统核心数据还得去业务系统库获取。
2、还有人问加入Kafka太多余
答：第一、加入Kafka为了使系统扩展性更强，可方便对接各种开源产品。第二、通过Kafka消息组可使同一条消息被不同Consumer消费，用户离线和实时两条线。
解析Mysql binlog日志
主要逻辑
1.创建Canal连接
2.解析Mysql binlog获得insert语句
public static void main(String args[]) {
//第一个参数为Canal server服务IP地址如果使用windows开发连接linux Canal服务需要制定IP eg: new InetSocketAddress("192.168.61.132", 11111)
//第二个参数为Canal server服务端口号 Canal server IP和端口号在 /conf/canal.properties中配置
//第三个参数为Canal instance名称 /conf下目录名称
//第四第五个参数为mysql用户名和密码，如果在 /conf/example/instance.properties中已经配置 这里不用谢
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); // 获取指定数量的数据
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); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
}
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
}
}
组装数据发送至Kafka
private static void printColumn(List columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue());
KafkaProducer.sendMsg("canal", UUID.randomUUID().toString() ,column.getName() + " : " + column.getValue());
}
}
Streaming分渠道汇总数据
以DStream中的数据进行按key做reduce操作，然后对各个批次的数据进行累加
在有新的数据信息进入或更新时，可以让用户保持想要的任何状。使用这个功能需要完成两步：
定义状态：可以是任意数据类型
定义状态更新函数：用一个函数指定如何使用先前的状态，从输入流中的新值更新状态。
对于有状态操作，要不断的把当前和历史的时间切片的RDD累加计算，随着时间的流失，计算的数据规模会变得越来越大。
val orders = resut_lines.updateStateByKey(updateRunningSum _)
def updateRunningSum(values: Seq[Long], state: Option[Long]) = {
/*
state:存放的历史数据
values:当前批次汇总值
*/
Some(state.getOrElse(0L)+values.sum)
}
统计结果写入Mysql
实时汇总某渠道下单量需要根据渠道为主键更新或插入新数据
1.当某个渠道第一单时，库中没有以此渠道为主键的数据，需要insert into 订单统计表
2.当某渠道在库中已有该渠道下单量，需要更新此渠道下单量值 update 订单统计表
所以我们使用：
#有该渠道就更新，没有就插入
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版本
com.google.protobuf
protobuf-java
2.4.1
参考文章
作者：MichaelFly
链接：https://www.jianshu.com/p/3ec093a9d584
來源：简书
简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。