背景
????????我们的埋点数据上传到S3，大概是每天10亿条的数据量级别。最近花了一些时间思考和学习如何将每天如此大量的数据从S3导入到Clickhouse，为后续的实时查询做准备。
方案一
? ? ? ? 1. 先将S3的数据导入到hive，这一步操作比较简单，创建一个外部表即可，按日期字段进行分区。
CREATE TABLE `s3_to_hive_test`(
id ,
aaa ,
bbb ,
ccc ,
ddd ,
…… )
PARTITIONED BY (
`ingestion_date` string)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
's3://host/bucket/path'
TBLPROPERTIES (
'last_modified_by'='hadoop',
'last_modified_time'='1625729438',
'parquet.column.index.access'='true',
'spark.sql.create.version'='2.2 or prior',
'spark.sql.sources.schema.numPartCols'='1',
'spark.sql.sources.schema.numParts'='1',
'spark.sql.sources.schema.part.0'='{……}',
'spark.sql.sources.schema.partCol.0'='ingestion_date')????????2. 每天定时从hive把前一天的数据导入到Clickhouse，这里可以借助waterdrop（现改名为seatunnel）工具进行导入，定时调度可以自己写一个shell脚本，如果导入后还需要进行数据清洗、聚合等，推荐DolphinScheduler。
#waterdrop的config文件
spark {
spark.app.name = "Waterdrop"
spark.executor.cores = 1
spark.executor.memory = "2g" // 这个配置必需填写，否则会使用 sparksql 内置元数据库
spark.default.parallelism = 12
spark.driver.memory = "64g"
spark.sql.catalogImplementation = "hive"
}
input {
hive {
pre_sql = "select `id`,`aaa`,`bbb`,`ccc`,`ddd`,…… from default.s3_to_hive_test where ……"
table_name = "waterdrop_tmp"
}
}
filter {
}
output {
clickhouse {
host = "ch_host:ch_port"
database = "default"
username = "***"
password = "******"
table = "ch_mergetree_test"
bulk_size = 200000
retry = 3
fields =
[ id ,
aaa ,
bbb ,
ccc ,
ddd ,
…… ]
}
}????????3. 导入后的清洗、聚合等操作。
适用场景：
? ? ? ? 绝大多数的场景都适用，之前我们项目也是采用此方案。目前由于引入K8S进行管理，hive、Spark、waterdrop、DolphinScheduler分属不同容器，要处理太多网络通信的问题，故改用更为简单的方案三。
方案二
? ? ? ? 1. 如方案一，先将S3的数据以外部表的方式导入到hive中并按日期分区。
CREATE TABLE `s3_to_hive_test`(
id ,
aaa ,
bbb ,
ccc ,
ddd ,
…… )
PARTITIONED BY (
`ingestion_date` string)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
's3://host/bucket/path'
TBLPROPERTIES (
'last_modified_by'='hadoop',
'last_modified_time'='1625729438',
'parquet.column.index.access'='true',
'spark.sql.create.version'='2.2 or prior',
'spark.sql.sources.schema.numPartCols'='1',
'spark.sql.sources.schema.numParts'='1',
'spark.sql.sources.schema.part.0'='{……}',
'spark.sql.sources.schema.partCol.0'='ingestion_date')? ? ? ? 2. Clickhouse建立hive集成表
CREATE TABLE IF NOT EXISTS hive_to_ch_test
(
`id` String NULL,
`aaa` String NULL,
`bbb` String NULL,
`ccc` String NULL,
`ddd` String NULL,
……
) ENGINE = Hive('thrift://host:port', 'database', 'table');
PARTITION BY ingestion_date
??????????3. Clickhouse内部将hive集成表的数据导入到MergeTree表
insert into ch_mergetree_test(
id ,
aaa ,
bbb ,
ccc ,
ddd ,
…… )
SELECT
id,
ifNull(aaa, ''),
ifNull(bbb, ''),
ifNull(ccc, ''),
ifNull(ddd, ''),
……
FROM hive_to_ch_test
WHERE ……
? ? ? ? 4. 后续的数据清洗、聚合等操作
????????在实测中，Clickhouse到22.4版本为止似乎还不支持hive集成表的底层存储为S3这种形式。具体表现为能建立hive集成表，但查询的时候报以下错误：
Query id: bfeb2774-eb2b-4b2c-9230-64bd6d35acfe
0 rows in set. Elapsed: 0.013 sec.
Received exception from server (version 22.4.5):
Code: 210. DB::Exception: Received from localhost:9000. DB::Exception: Unable to connect to HDFS: InvalidParameter: Cannot parse URI: hdfs://****, missing port or invalid HA configuration Caused by: HdfsConfigNotFound: Config key: dfs.ha.namenodes.**** not found. (NETWORK_ERROR)
但支持底层为HDFS，如下图所示。（注意：如HDFS是HA模式，需要参考官方文档进行一些额外的配置，否则也会报以上错误）
?适用场景：
? ? ? ? Clickhouse版本必须高于22.1，因为hive集成表引擎是在22.1版本才发布的。详见Clickhouse ChangeLog
方案三
?大道至简，直接省略S3导入到Hive这个步骤。
? ? ? ? 1. 建立S3集成表
DROP TABLE IF EXISTS s3_to_ch_test ;
CREATE TABLE s3_to_ch_test (
`id` String NULL,
`aaa` String NULL,
`bbb` String NULL,
`ccc` String NULL,
`ddd` String NULL,
…… )
ENGINE=S3(concat('https://s3_host/bucket/path/*' ),'accessKey','secretKey', 'Parquet')
SETTINGS input_format_parquet_allow_missing_columns=true ;? ? ? ? 2. Clickhouse内部将数据从S3集成表导入MergeTree表
insert into ch_mergetree_test(
id ,
aaa ,
bbb ,
ccc ,
ddd ,
…… )
SELECT
id,
ifNull(aaa, ''),
ifNull(bbb, ''),
ifNull(ccc, ''),
ifNull(ddd, ''),
……
FROM s3_to_ch_test
WHERE ……? ? ? ? 3. 后续的数据清洗、集成等操作
????????当然，在具体实施过程，还是会遇到一些坑。
坑一、超出内存限制
????????由于我们在S3中存储的文件格式是Parquet类型，Parquet是面向分析型业务的列式存储格式，Clickhouse在处理Parquet文件是内存密集型的。我尝试将output_format_parquet_row_group_size?参数调小，但没有任何作用，仍报上述异常。猜测Clickhouse在查询数据时并不是按文件一个个读取并插入的，而是将所有文件的列为单位装到内存中。因为我单个文件最大也才10G，分配了50G内存仍超出内存。
解决方案：调大内存限制的值。
????????方式1. 临时设置（仅对当前session有效）：
set max_memory_usage=150000000000;
select * from system.settings where name='max_memory_usage';????????方式2. 修改/etc/clickhouse-server/users.xml文件（长期有效）
<max_memory_usage>150000000000</max_memory_usage>坑二、效率问题
解决方案1：加大内存
????????内存100G的情况下，实测10亿条数据导入到MergeTree表消耗时间4300+ Sec，加到150G消耗时间3800+ Sec。
解决方案2：增加线程数
????????当内存150G，max_insert_threads=1时，耗时3837s
?????????当内存150G，max_insert_threads=2时，耗时1939s
????????当内存150G，max_insert_threads=3时，耗时1407s
????????当内存150G，max_insert_threads=4时,超出内存限制！！
????????虽然增加线程数对导入效率会有明显提升，但不意味着线程数越多越好，因为高线程数是以高内存为代价的，需要根据服务器内存和导入数据量的情况，平衡好max_memory_usage和max_insert_threads的关系。?