ClickHouse学习系列之⼋【数据导⼊迁移同步】
背景
在介绍了⼀些ClickHouse相关的之后，⼤致对ClickHouse有了⽐较多的了解。它是⼀款⾮常优秀的OLAP数据库，为了更好的来展⽰其强⼤的OLAP能⼒，本⽂将介绍⼀些快速导⼊⼤量数据到ClickHouse的⽅法。如：通过⽂件、远程数据库等⽅式。
说明
⼀、其他数据库导⼊到ClickHouse
MySQL可以做为ClickHouse的外部存储类型，还有其他的存储类型，如：MongoDB、PostgreSQL、HDFS、JDBC、ODBC、Kafka、File、RabbitMQ、S3等等，具体的可以看。本⽂介绍MySQL、MongoDB、File三种⽅式的导⼊，前2种⽅式相当于链表。
①
⽅法⼀：外部引擎，建⽴远程表
MySQL引擎允许对存储在远程 MySQL 服务器上的数据执⾏ SELECT 和 INSERT 查询，不能执⾏DELETE 和 UPDATE。
CREATE TABLE[IF NOT EXISTS][db.]table_name [ON CLUSTER cluster]
(
name1 [type1][DEFAULT|MATERIALIZED|ALIAS expr1][TTL expr1],
name2 [type2][DEFAULT|MATERIALIZED|ALIAS expr2][TTL expr2],
...
) ENGINE = MySQL('host:port', 'database', 'table', 'user', 'password'[, replace_query, 'on_duplicate_clause'])
SETTINGS
[connection_pool_size=16, ]
[connection_max_tries=3, ]
[connection_auto_close=true ]
;
建议：列名相同。参数表⽰如何处理null值，默认1，可选值0。1：可以为null，0：不能为null，使⽤默认值代替。MySQL的参数说明：
host:port — MySQL 地址
database — MySQL数据库名
table — MySQL表名
user — MySQL ⽤户
password — MySQL⽤户密码
replace_query — 默认0，对应replace into。设置1这会⽤replace into 代替 insert into
on_duplicate_clause — 默认0，对应 ON DUPLICATE KEY。设置1这会代替 insert into，和replace_query互斥
例⼦：
CREATE st
(
`id` UInt32,
`c1` Date COMMENT 'c1',
`c2` DateTime COMMENT 'c2',
`c3` String COMMENT 'c3',
`c4` UInt32 COMMENT 'c4'
)
ENGINE = MySQL('10.10.10.10:3306','test','test','dba','dba')
注意：在查询远程表的时候，除了where条件会带⼊到远程的MySQL中，其余的条件（聚合），包含li
mit都会在ClickHouse本地执⾏，⽽远程则执⾏全表扫描。最后可以对该表执⾏SELECT 和 INSERT 查询，不能执⾏DELETE 和 UPDATE。
此后就可以在ClickHouse上创建符合要求的引擎表，如MergeTree引擎，再通过以下SQL来进⾏导⼊数据。
insert into ck_tb select*from my_tb
⽅法⼆：和⽅法⼀类似，不过⽅法⼆是直接⽤了mysql函数来进⾏远程访问：建⽴需要的引擎，再⽤mysql函数进⾏远程导⼊，可以指定列来导⼊数据，把*改成具体列名。
insert into ck_tb select*from mysql('host:port', 'database', 'table', 'user', 'password')
mysql函数⾥的参数可以参考⽅法⼀的说明。
例⼦：
-- 建⽴表
CREATE st_ck1
(
`id` UInt32,
`c1` Date COMMENT 'c1',
`c2` DateTime COMMENT 'c2',
`c3` String COMMENT 'c3',
`c4` UInt32 COMMENT 'c4'
)
ENGINE = MergeTree
PARTITION BY c1
ORDER BY id
-- 插⼊数据
INSERT INTO test_ck1 SELECT*
FROM mysql('10.10.10.10:3306','test','test','dba','dba')
⽅法三：create + select
CREATE TABLE[IF NOT EXISTS][db.]table_name
ENGINE = ENGINE
AS
SELECT*
FROM mysql('host:port', 'database', 'table', 'user', 'password')
mysql函数⾥的参数可以参考⽅法⼀的说明。
例⼦：
create table test_ck2 engine = MergeTree order by id as select*from mysql('10.10.10.10:3306','test','test','dba','dba');
或
create table test_ck2 engine =Log as select*from mysql('10.10.10.10:3306','test','test','dba','dba');
通过该⽅法的导⼊，不需要事先建⽴表，它会⾃动根据select出来的数据按照需要创建符合要求的类型。
②
MongoDB引擎允许对存储在远程 MongoDB 服务器上的数据执⾏ SELECT 查询，不能执⾏ INSERT、DELETE 和 UPDATE。
CREATE TABLE[IF NOT EXISTS][db.]table_name
(
name1 [type1],
name2 [type2],
...
) ENGINE = MongoDB(host:port, database, collection, user, password);
建议：列名相同。MongoDB的参数说明：
host:port — MongoDB 地址.
database — MongoDB 数据库名
collection — MongoDB 集合名
user — MongoDB ⽤户
password — MongoDB 密码
例⼦：
CREATE TABLE mongo_table
(
key UInt64,
data String
) ENGINE = MongoDB('mongo1:27017', 'test', 'simple_table', 'testuser', 'clickhouse');
注意：相对于MySQL，MongoDB远程表还不允许INSERT。此后就可以在ClickHouse上创建符合要求的引擎表，如MergeTree引擎，再通过以下SQL来进⾏导⼊数据。
insert into ck_tb select*from mon_tb
MongoDB⽬前只作为⼀个外部引擎，不像MySQL还能作为⼀个函数进⾏远程操作，所以MySQL数据处理中的⽅法⼆、三不适⽤于MongoDB数据的处理。
③：和
和MySQL引擎导⼊⽅法类似，⽀持并发读，不⽀持并发插⼊。不⽀持ALTER、索引和副本
⽅法⼀：（建⽴远程表）
CREATE TABLE file_engine_table (name String, value UInt32) ENGINE=File(Format)
File函数⾥的参数的取值可以看，这⾥说明下csv的例⼦。
-- 建⽴表
create table csv_table(id UInt64,name String)engine =File('CSV');
-- 在表⽬录⾥创建⽂件或则导⼊⽂件，必须命名为data.CSV
$ cat data.CSV
1,a
2,b
3,c
4,d
5,e
-- 这样，在表⾥就可以看到数据了
:) select*from csv_table;
┌─id─┬─name─┐
│1│ a │
│2│ b │
│3│ c │
│4│ d │
│5│ e │
└────┴──────┘
注意：File引擎的表，可以对其进⾏SELECT、INSERT，不能进⾏DELETE、UPDATE。此后就可以在ClickHouse上创建符合要求的引擎表，如MergeTree引擎，再通过以下SQL来进⾏导⼊数据。
insert into ck_tb select*from csv_tb
⽅法⼆：
通过file函数导⼊数据，并且以表的形式展⽰，格式为：
select * from file(path, format, structure)
path — 参数下的相对路径，⽀持以下格式：*、?、{abc,def} 和 {N..M} ，其中 N、M — 数字、'abc'、'def' — 字符串。
format —
structure — 表结构。格式：'column1_name column1_type, column2_name column2_type, ...'
在l⽂件中到参数，在该参数指定的⽬录下创建⼀个csv⽂件：
$ cat test.csv
1,a,123
2,b,234
3,c,345
4,d,456
5,e,567
然后通过SQL查询：
:) SELECT*FROM file('test.csv', 'CSV', 'column1 UInt32, column2 String, column3 UInt32');
┌─column1─┬─column2─┬─column3─┐
│1│ a │123│
│2│ b │234│
│3│ c │345│
│4│ d │456│
│5│ e │567│
└─────────┴─────────┴─────────┘
如果该⽬录下有多个csv⽂件：test.csv、test1.csv、test2.csv，则可以通过通配符来进⾏全部加载读取：
:) SELECT*FROM file('test*.csv', 'CSV', 'column1 UInt32, column2 String, column3 UInt32');
┌─column1─┬─column2─┬─column3─┐
│1│ aaa │123│
│2│ bbb │234│
│3│ ccc │345│
│4│ ddd │456│
│5│ eee │567│
└─────────┴─────────┴─────────┘
┌─column1─┬─column2─┬─column3─┐
│1│ a │123│
│2│ b │234│
│3│ c │345│
│4│ d │456│
│5│ e │567│
└─────────┴─────────┴─────────┘
┌─column1─┬─column2─┬─column3─┐
│1│ aa │123│
│2│ bb │234│
│3│ cc │345│
│4│ dd │456│
│5│ ee │567│
└─────────┴─────────┴─────────┘
此后就可以在ClickHouse上创建符合要求的引擎表，如MergeTree引擎，再通过以下SQL来进⾏导⼊数据。
:) insert into csv_table_ck SELECT*FROM file('test*.csv', 'CSV', 'column1 UInt32, column2 String, column3 UInt32')
⽅法三：create + select
CREATE TABLE[IF NOT EXISTS][db.]table_name
ENGINE = ENGINE
AS
SELECT*
FROM file(path, format, structure)
例⼦：
:) create table csv_table_ck1 engine = MergeTree ORDER BY column1 as SELECT*FROM file('test*.csv', 'CSV', 'column1 UInt32, column2 String, column3 UInt32');
注意：需要在file函数的参数structure指定的表结构字段中选取字段进⾏ ORDER BY 的指定设置。并
且通过该⽅法的导⼊，不需要事先建⽴表，它会⾃动根据select出来的数据按照需要创建符合要求的类型。
⽅法四：也可以通过输⼊流来进⾏数据的导⼊，
$ echo -e "1,2\n3,4" | clickhouse-local -q "CREATE TABLE table (a Int64, b Int64) ENGINE =File(CSV, stdin); SELECT a, b FROM table; DROP TABLE table"
⼆、（实验阶段）
ClickHouse推出了数据库引擎，⽤于将MySQL服务器中的表映射到ClickHouse中。ClickHouse服务做为MySQL副本，读取Binlog并执⾏DDL和DML请求，实现了基于MySQL Binlog机制的业务数据库实时同步功能：⽀持全量和增量同步，⾸次创建数据库引擎时进⾏⼀次全量复制，之后通过监控binlog变化进⾏增量数据同步。阿⾥云上也有。
CREATE DATABASE[IF NOT EXISTS]db_name[ON CLUSTER cluster]
ENGINE = MaterializeMySQL('host:port', ['database' | database], 'user', 'password') [SETTINGS ...]
host:port — MySQL 地址
database — MySQL 数据库名
user — MySQL ⽤户名，具有MySQL库的RELOAD、REPLICATION SLAVE、REPLICATION CLIENT以及SELECT PRIVILEGE权限
password — MySQL 密码
使⽤数据库引擎，同步到ClickHouse集上表的默认引擎为ReplacingMergeTree，并会在表中增加2个虚拟列：
_version — 事务计数器，记录数据版本信息。UInt64类型。
_sign — 删除标记，标记该⾏是否删除。TypeInt8类型:
1 — 未删除
-1 — 已删除
测试：
数据库引擎需要开启allow_experimental_database_materialize_mysql参数。即需要设置为1：
SET allow_experimental_database_materialize_mysql =1
以上SET 只是更改了当前会话中的值，分布式 DDL 在单独的会话中执⾏，SET 不影响它。应该在服务器配置中全局启⽤ allow_experimental_database_materialize_mysql 设置，如：
修改l：
<profiles>
<!-- Default settings. -->
<default>
...
<allow_experimental_database_materialize_mysql>1</allow_experimental_database_materialize_mysql>
</default>
...
</profiles>
修改l：
<distributed_ddl>
...
<path>/clickhouse/task_queue/ddl</path>
<allow_experimental_database_materialize_mysql>1</allow_experimental_database_materialize_mysql>
...
</distributed_ddl>
修改这些配置是动态⽣效的，可以查看该参数是否修改成功：
:) SELECT* FROM system.settings WHERE name ='allow_experimental_database_materialize_mysql';
┌─name──────────────────────────────────────────┬─value─┬─changed─┬─description─────────────────────────────────────────────────┬─min──┬─max──┬─readonly─┬─type─┐
│ allow_experimental_database_materialize_mysql │1│1│ Allow to create database with Engine=MaterializeMySQL(...). │ᴺᵁᴸᴸ│ᴺᵁᴸᴸ│0│ Bool │
└───────────────────────────────────────────────┴───────┴─────────┴─────────────────────────────────────────────────────────────┴──────┴──────┴──────────┴──────┘现在可以开始数据库引擎来同步MySQL数据库了。需要注意的是MySQL的binlog格式为Row，并且开启GTID。
①：创建数据库：同步MySQL的ck_test库中的表
:) create database mysql_2_ck ENGINE = MaterializeMySQL('10.10.10.10:3306','ck_test','test','test');
②：全量同步
-- 全量：
通过select*from tb 拉取MySQL的全量数据进⾏同步，再在本地ClickHouse上回放
③：增量同步
-- 增量：
通过订阅MySQL binlog来进⾏增量同步，在mysql上可以看到 Binlog Dump GTID 的订阅线程，再在本地ClickHouse上回放
④：记录MySQL Binlog信息
-- 记录 binlog 信息
在<path>指定的⽬录下的metadata/dbname中保存binlog和position：<path>/metadata/dbname/.metadata
通过①~⑤同步关系已经搭建完成，为了测试同步效果，进⾏测试：测试同步包括：insert，update，delete，alter，create，drop，truncate等⼤部分DML和DDL操作
说明：在①中已经建⽴了ClickHouse【mysql_2_ck】和MySQL【ck_test】库的同步，后续操作在该库中进⾏。
新建表
-- mysql> CREATE TABLE employees (
-> emp_no INT NOT NULL,
-> birth_date DATE NOT NULL,
-> first_name VARCHAR(14) NOT NULL,
-> last_name VARCHAR(16) NOT NULL,
-> gender CHAR(3) NOT NULL,
-> hire_date DATE NOT NULL,
->PRIMARY KEY (emp_no)
-> ) ENGINE=INNODB;
-- clickhouse :) show create table employees\G
statement: CREATE TABLE mysql_ployees
(
`emp_no` Int32,
`birth_date` Date,
`first_name` String,
`last_name` String,
`gender` String,
`hire_date` Date,
`_sign` Int8 MATERIALIZED 1,
`_version` UInt64 MATERIALIZED 1,
INDEX _version _version TYPE minmax GRANULARITY 1
)
ENGINE = ReplacingMergeTree(_version)
PARTITION BY intDiv(emp_no, 4294967)
ORDER BY tuple(emp_no)
SETTINGS index_granularity =8192
新增数据
-- mysql> INSERT INTO `employees` VALUES (10001,'1953-09-02','Georgi','Facello','M','1986-06-26'),
-> (10002,'1964-06-02','Bezalel','Simmel','F','1985-11-21'),
-> (10003,'1959-12-03','Parto','Bamford','M','1986-08-28'),
-> (10004,'1954-05-01','Chirstian','Koblick','M','1986-12-01'),
-> (10005,'1955-01-21','Kyoichi','Maliniak','M','1989-09-12'),
-> (10006,'1953-04-20','Anneke','Preusig','F','1989-06-02'),
-> (10007,'1957-05-23','Tzvetan','Zielinski','F','1989-02-10'),
-> (10008,'1958-02-19','Saniya','Kalloufi','M','1994-09-15'),
-> (10009,'1952-04-19','Sumant','Peac','F','1985-02-18'),
-> (10010,'1963-06-01','Duangkaew','Piveteau','F','1989-08-24'),
-> (10011,'1953-11-07','Mary','Sluis','F','1990-01-22'),
-> (10012,'1960-10-04','Patricio','Bridgland','M','1992-12-18'),
-> (10013,'1963-06-07','Eberhardt','Terkki','M','1985-10-20'),
-> (10014,'1956-02-12','Berni','Genin','M','1987-03-11'),
-> (10015,'1959-08-19','Guoxiang','Nooteboom','M','1987-07-02'),
-> (10016,'1961-05-02','Kazuhito','Cappelletti','M','1995-01-27'),
-> (10017,'1958-07-06','Cristinel','Bouloucos','F','1993-08-03'),
-> (10018,'1954-06-19','Kazuhide','Peha','F','1987-04-03');
-- clickhouse :) select * from mysql_ployees;
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name───┬─gender─┬──hire_date─┐
│10002│2143-11-07│ Bezalel │ Simmel │ F │1985-11-21│
│10006│2132-09-24│ Anneke │ Preusig │ F │1989-06-02│
│10011│2133-04-13│ Mary │ Sluis │ F │1990-01-22│
│10015│2139-01-23│ Guoxiang │ Nooteboom │ M │1987-07-02│
│10016│2140-10-06│ Kazuhito │ Cappelletti │ M │1995-01-27│
└────────┴────────────┴────────────┴─────────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10004│2133-10-05│ Chirstian │ Koblick │ M │1986-12-01│
│10009│2131-09-24│ Sumant │ Peac │ F │1985-02-18│
│10013│2142-11-11│ Eberhardt │ Terkki │ M │1985-10-20│
│10018│2133-11-23│ Kazuhide │ Peha │ F │1987-04-03│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10003│2139-05-09│ Parto │ Bamford │ M │1986-08-28│
│10007│2136-10-27│ Tzvetan │ Zielinski │ F │1989-02-10│
│10010│2142-11-05│ Duangkaew │ Piveteau │ F │1989-08-24│
│10014│2135-07-19│ Berni │ Genin │ M │1987-03-11│
│10017│2137-12-10│ Cristinel │ Bouloucos │ F │1993-08-03│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10001│2133-02-06│ Georgi │ Facello │ M │1986-06-26│
│10005│2134-06-27│ Kyoichi │ Maliniak │ M │1989-09-12│
│10008│2137-07-26│ Saniya │ Kalloufi │ M │1994-09-15│
│10012│2140-03-10│ Patricio │ Bridgland │ M │1992-12-18│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
删除数据
-- mysql> delete from employees where emp_no >10010;
Query OK, 8 rows affected (0.01 sec)
-- clickhouse :) select * from mysql_ployees;
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10001│2133-02-06│ Georgi │ Facello │ M │1986-06-26│
│10005│2134-06-27│ Kyoichi │ Maliniak │ M │1989-09-12│
│10008│2137-07-26│ Saniya │ Kalloufi │ M │1994-09-15│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10004│2133-10-05│ Chirstian │ Koblick │ M │1986-12-01│
│10009│2131-09-24│ Sumant │ Peac │ F │1985-02-18│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10003│2139-05-09│ Parto │ Bamford │ M │1986-08-28│
│10007│2136-10-27│ Tzvetan │ Zielinski │ F │1989-02-10│
│10010│2142-11-05│ Duangkaew │ Piveteau │ F │1989-08-24│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10002│2143-11-07│ Bezalel │ Simmel │ F │1985-11-21│
│10006│2132-09-24│ Anneke │ Preusig │ F │1989-06-02│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
修改数据
-- mysql> update employees set hire_date = hire_date+1 where emp_no <10005;
Query OK, 4 rows affected (0.01 sec)
-- clickhouse :) select * from mysql_ployees;
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10001│2133-02-06│ Georgi │ Facello │ M │1986-06-27│
│10005│2134-06-27│ Kyoichi │ Maliniak │ M │1989-09-12│
│10008│2137-07-26│ Saniya │ Kalloufi │ M │1994-09-15│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10003│2139-05-09│ Parto │ Bamford │ M │1986-08-29│
│10007│2136-10-27│ Tzvetan │ Zielinski │ F │1989-02-10│
│10010│2142-11-05│ Duangkaew │ Piveteau │ F │1989-08-24│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10002│2143-11-07│ Bezalel │ Simmel │ F │1985-11-22│
│10006│2132-09-24│ Anneke │ Preusig │ F │1989-06-02│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
┌─emp_no─┬─birth_date─┬─first_name─┬─last_name─┬─gender─┬──hire_date─┐
│10004│2133-10-05│ Chirstian │ Koblick │ M │1986-12-02│
│10009│2131-09-24│ Sumant │ Peac │ F │1985-02-18│
└────────┴────────────┴────────────┴───────────┴────────┴────────────┘
修改表结构
-新增字段
-- mysql> alter table employees add age int after gender;
mongodb和mysql结合Query OK, 0 rows affected (0.08 sec)
-- clickhouse :) desc mysql_ployees;
┌─name───────┬─type────────────┬─default_type─┬─default_expression─┬─comment─┬─codec_expression─┬─ttl_expression─┐
│ emp_no │ Int32 ││││││
│ birth_date │ Date ││││││
│ first_name │ String ││││││
│ last_name │ String ││││││
│ gender │ String ││││││
│ age │ Nullable(Int32) ││││││
│ hire_date │ Date ││││││
│ _sign│ Int8 │ MATERIALIZED │1││││
│ _version │ UInt64 │ MATERIALIZED │1││││
└────────────┴─────────────────┴──────────────┴────────────────────┴─────────┴──────────────────┴────────────────┘-删除字段
-- mysql> alter table employees drop age;
Query OK, 0 rows affected (0.09 sec)
-- clickhouse :) desc mysql_ployees;
┌─name───────┬─type───┬─default_type─┬─default_expression─┬─comment─┬─codec_expression─┬─ttl_expression─┐
│ emp_no │ Int32 ││││││
│ birth_date │ Date ││││││
│ first_name │ String ││││││
│ last_name │ String ││││││
│ gender │ String ││││││
│ hire_date │ Date ││││││
│ _sign│ Int8 │ MATERIALIZED │1││││
│ _version │ UInt64 │ MATERIALIZED │1││││
└────────────┴────────┴──────────────┴────────────────────┴─────────┴──────────────────┴────────────────┘
新增索引
-- ClickHouse不会同步
删除索引
-- ClickHouse不会同步
修改字段长度
-加长
-- mysql> alter table employees modify age bigint;
-- clickhouse :) desc employees;
┌─name───────┬─type─────────────┬─default_type─┬─default_expression─┬─comment─┬─codec_expression─┬─ttl_expression─┐
│ emp_no │ Int32 ││││││
│ birth_date │ Date ││││││
│ first_name │ String ││││││
│ last_name │ String ││││││
│ gender │ String ││││││
│ age │ Nullable(Int64) ││││││
│ address │ Nullable(String) ││││││
│ hire_date │ Date ││││││
│ _sign│ Int8 │ MATERIALIZED │1││││
│ _version │ UInt64 │ MATERIALIZED │1││││
└────────────┴──────────────────┴──────────────┴────────────────────┴─────────┴──────────────────┴────────────────┘-改短
-- mysql> alter table employees modify age int;
-- clickhouse :) desc employees;
┌─name───────┬─type─────────────┬─default_type─┬─default_expression─┬─comment─┬─codec_expression─┬─ttl_expression─┐
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论