python 利用binlog实时计算

因为MySQL本身就有帮你记录日志, 而且记录的日志比自己应用程序的要准确的多(MySQL 需要开启row模式)。
解析MySQL Binlog，并获取我们想要的Event。
过滤出我们关心的表。
将获得的相关数据实时的保存到相关存储中(一般使用redis存数据，之后再定时刷入MySQL)。
模拟步骤：
我们这边模拟实时算订销售总额，和订单量。
使用python-mysql-replication作为实时解析MySQL Binlog的日志利器(推荐使用阿里的 canal，这里主要看公司的开发人员擅长什么而决定)。
我们只关心 WriteRowsEvent (事件号 30)。
我们只关心 ord_order 表产生的 WriteRowsEvent 事件。
在原来统计的基础上加上本次订单的信息并保存到Redis(使用打印来代替保存到Redis)。
老套的 SQL 代码

–创建演示的 ord_order 表
CREATE TABLE ord_order(
order_id INT NOT NULL AUTO_INCREMENT COMMENT ‘订单ID’,
amount INT NOT NULL DEFAULT 0 COMMENT ‘订单金额(分)’,
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT ‘创建时间’,
PRIMARY KEY(order_id)
)COMMENT = ‘订单表’;

— 插入几笔订单
INSERT INTO ord_order(amount) VALUES
(1000),
(2000),
(3000);

— 插入几笔订单
INSERT INTO ord_order(amount) VALUES
(1000),
(2000),
(3000);

#!/usr/bin/env python
# -*- coding:utf-8 -*-

from pymysqlreplication import BinLogStreamReader

# 链接数据库的参数因为 pymysqlreplication 底层使用的是 python-mysql
mysql_settings = {
‘host’: ‘192.168.1.233’,
‘port’: 3306,
‘user’: ‘HH’,
‘passwd’: ‘oracle‘
}

# 这 pymysqlreplication 的 server_id 和从 Binlog 的什么位置开始解析
stream = BinLogStreamReader(connection_settings=mysql_settings,
server_id=100,
blocking=True,
log_file=’mysql-bin.000012’,
log_pos=469)

# 初始化订单统计数据
order_count_total = 0
order_amount_total = 0

# 不停的解析获取解析的 Binlog
for binlogevent in stream:

# 碰到 WriteRowsEvent 并且表是 ord_order 则进行统计
if binlogevent.event_type == 30 and binlogevent.table == ‘ord_order’:
binlogevent.dump() # 打印事件相关信息

# 同时计算出订单数和金额数组, 如: [(1, 9), (1, 4)]
stat = [(1, row[‘values’][‘amount’]) for row in binlogevent.rows]

# 分别获得订单数数组如:[1, 1]。销售额, 如: [9, 4]
order_count, order_amount = zip(*stat)
order_count_total += sum(order_count)
order_amount_total += sum(order_amount)

# 打印本次事件产生的订单数和销售额
print ‘order_count:’, order_count
print ‘order_amount:’, order_amount

# 打印总的订单数和销售额
print ‘order_count_total:’, order_count_total
print ‘order_amount_total:’, order_amount_total

#!/usr/bin/env python
# -*- coding:utf-8 -*-

from pymysqlreplication import BinLogStreamReader

# 链接数据库的参数因为 pymysqlreplication 底层使用的是 python-mysql
mysql_settings = {
‘host’: ‘192.168.1.233’,
‘port’: 3306,
‘user’: ‘HH’,
‘passwd’: ‘oracle’
}

# 初始化订单统计数据
order_count_total = 0
order_amount_total = 0

# 不停的解析获取解析的 Binlog
for binlogevent in stream:

# 碰到 WriteRowsEvent 并且表是 ord_order 则进行统计
if binlogevent.event_type == 30 and binlogevent.table == ‘ord_order’:
binlogevent.dump() # 打印事件相关信息

# 同时计算出订单数和金额数组, 如: [(1, 9), (1, 4)]
stat = [(1, row[‘values’][‘amount’]) for row in binlogevent.rows]

# 分别获得订单数数组如:[1, 1]。销售额, 如: [9, 4]
order_count, order_amount = zip(*stat)
order_count_total += sum(order_count)
order_amount_total += sum(order_amount)

# 打印本次事件产生的订单数和销售额
print ‘order_count:’, order_count
print ‘order_amount:’, order_amount

# 打印总的订单数和销售额
print ‘order_count_total:’, order_count_total
print ‘order_amount_total:’, order_amount_total
运行代码

[root@centos7 tmp]# python test.py
=== WriteRowsEvent ===
Date: 2016-11-16T17:11:11
Log position: 681
Event size: 54
Read bytes: 12
Table: test.ord_order
Affected columns: 3
Changed rows: 3
Values:
—
(‘*’, u’order_id’, ‘:’, 1)
(‘*’, u’amount’, ‘:’, 1000)
(‘*’, u’create_time’, ‘:’, datetime.datetime(2016, 11, 16, 17, 11, 11))
—
(‘*’, u’order_id’, ‘:’, 2)
(‘*’, u’amount’, ‘:’, 2000)
(‘*’, u’create_time’, ‘:’, datetime.datetime(2016, 11, 16, 17, 11, 11))
—
(‘*’, u’order_id’, ‘:’, 3)
(‘*’, u’amount’, ‘:’, 3000)
(‘*’, u’create_time’, ‘:’, datetime.datetime(2016, 11, 16, 17, 11, 11))
()
order_count: (1, 1, 1)
order_amount: (1000, 2000, 3000)
order_count_total: 3
order_amount_total: 6000

ALTER TABLE ord_order
ADD PARTITION (PARTITION p201701 VALUES IN (201701));

ALTER TABLE ord_order DROP PARTITION p201601;
1
2
3
4
ALTER TABLE ord_order
ADD PARTITION (PARTITION p201701 VALUES IN (201701));

ALTER TABLE ord_order DROP PARTITION p201601;
MySQL 挂掉要如何处理。
如何实现程序的高可用。
如何记录解析的 log file 和 log pos。
需不需要将解析的数据统一管理和存储。

python 利用binlog实时计算

One thought on “python 利用binlog实时计算”

发表评论取消回复

One thought on “python 利用binlog实时计算”

发表评论 取消回复

发表评论取消回复