Maxwell 系列（四）—— bootstrap数据全量导入

大数据最后一公里

发布于 2021-08-05 10:26:54

4K0

发布于 2021-08-05 10:26:54

文章被收录于专栏：大数据最后一公里

1、bootstrap使用

Maxwell允许您将数据“引导”到流中。这将执行 select * from table和将结果输出到您的流中，从而允许您从头开始播放流来重新创建整个数据集。

binlog被清除

maxwell是可以记录binlog读取的position，如果binlog被清除，则position无意义。

数据初始化

项目刚启动，需要把已经存在的历史数据同步到流中，然后再使用增量的方式抽取。

您可以使用该maxwell-bootstrap实用程序从命令行开始boostrap操作。

option	description
--log_level LOG_LEVEL	log level (DEBUG, INFO, WARN or ERROR)
--user USER	mysql username
--password PASSWORD	mysql password
--host HOST	mysql host
--port PORT	mysql port
--database DATABASE	mysql database containing the table to bootstrap
--table TABLE	mysql table to bootstrap
--where WHERE_CLAUSE	where clause to restrict the rows bootstrapped from the specified table
--client_id CLIENT_ID	specify which maxwell instance should perform the bootstrap operation
--comment COMMENT	arbitrary comment to be added to every bootstrap row record

选项	描述
--log_level LOG_LEVEL	日志级别（调试，信息，警告或错误）
--user USER	mysql用户名
--password密码	mysql密码
-主机主机	mysql主机
-端口	mysql端口
-数据库数据库	mysql数据库包含要引导的表
--table表	引导mysql表
-WHERE_CLAUSE	where子句限制从指定表引导的行
--client_id CLIENT_ID	指定哪个maxwell实例应执行引导操作
-评论评论	要添加到每个引导行记录的任意注释

启动表引导

bin/maxwell-bootstrap --database fooDB --table barTable

Maxwell复制器是单线程的；一个事件从binlog中捕获一个事件，一次将一条消息复制到Kafka。使用时运行Maxwell时--bootstrapper=sync，使用同一线程进行引导，这意味着所有binlog事件都将被阻止，直到引导完成。--bootstrapper=async但是，如果使用Maxwell 运行，将使Maxwell产生一个用于引导的单独线程。在这种异步模式下，非引导表将由主线程正常复制，而引导表的binlog事件将排队，并在引导过程结束时发送到复制流。

如果Maxwell在下次引导时崩溃，它将完全重新引导该引导程序-不管之前的进度如何。如果不需要此行为，则需要手动更新bootstrap表。具体来说，将未完成的引导程序行标记为“完成”（is_complete= 1）或删除该行。

bin/maxwell \
--user maxwell \
--password 123456 \
--host 10.100.97.246 \
--database test \
--table test \
--client_id maxwell

2、数据格式

mysql操作

create table test.e (
  id int(10) not null primary key auto_increment,
  m double,
  c timestamp(6),
  comment varchar(255) charset 'latin1'
);

insert into test.e set m = 4.2341, c = now(3), comment = 'I am a creature of light.';
update test.e set m = 5.444, c = now(3) where id = 1;
delete from test.e where id = 1;
alter table test.e add column torvalds bigint unsigned after m;
drop table test.e;

INSERT

mysql> insert into test.e set m = 4.2341, c = now(3), comment = 'I am a creature of light.';
{
   "database":"test",
   "table":"e",
   "type":"insert",
   "ts":1477053217,
   "xid":23396,
   "commit":true,
   "position":"master.000006:800911",
   "server_id":23042,
   "thread_id":108,
   "primary_key": [1, "2016-10-21 05:33:37.523000"],
   "primary_key_columns": ["id", "c"],
   "data":{
      "id":1,
      "m":4.2341,
      "c":"2016-10-21 05:33:37.523000",
      "comment":"I am a creature of light."
   }
}

UPDATE

mysql> update test.e set m = 5.444, c = now(3) where id = 1;
{
   "database":"test",
   "table":"e",
   "type":"update",
   "ts":1477053234,
   ...
   "data":{
      "id":1,
      "m":5.444,
      "c":"2016-10-21 05:33:54.631000",
      "comment":"I am a creature of light."
   },
   "old":{
      "m":4.2341,
      "c":"2016-10-21 05:33:37.523000"
   }
}

DELETE

mysql> delete from test.e where id = 1;
{
   "database":"test",
   "table":"e",
   "type":"delete",
   ...
   "data":{
      "id":1,
      "m":5.444,
      "c":"2016-10-21 05:33:54.631000",
      "comment":"I am a creature of light."
   }
}

关注公众号：大数据最后一公里

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-07-25，如有侵权请联系 cloudcommunity@tencent.com 删除

bootstrap