Clickhouse导数工具waterdrop用法

官方文档链接地址：https://interestinglab.github.io/waterdrop-docs/#/zh-cn/case_study/2
Hive to ClickHouse

假定我们的数据已经存储在Hive中，我们需要读取Hive表中的数据并筛选出我们关心的字段，或者对字段进行转换，最后将对应的字段写入ClickHouse的表中。

Hive Schema

我们在Hive中存储的数据表结构如下，存储的是很常见的Nginx日志
hive的建表语句如下：

CREATE TABLE `nginx_msg_detail` (
`hostname` string,
`domain` string,
`remote_addr` string,
`request_time` FLOAT,
`datetime` string,
`url` string,
`status` INT,
`data_size` INT,
`referer` string,
`cookie_info` string,
`user_agent` string,
`minute` string
) PARTITIONED BY (`date` string, `hour` string)

ClickHouse Schema

我们的ClickHouse建表语句如下，我们的表按日进行分区

CREATE TABLE cms.cms_msg (
date Date,
datetime DateTime,
url String,
request_time Float32,
STATUS String,
hostname String,
domain String,
remote_addr String,
data_size Int32
) ENGINE = MergeTree PARTITION BY date
ORDER BY
(date, hostname) SETTINGS index_granularity = 16384

Waterdrop with ClickHouse

接下来通过Waterdrop将Hive中的数据写入ClickHouse中。
Waterdrop是通过spark引擎来进行数据导入（spark-sql）
Waterdrop Pipeline
我们仅需要编写一个Waterdrop Pipeline的配置文件即可完成数据的导入。
配置文件包括四个部分，分别是Spark、Input、filter和Output。

在Waterdop的conf目录下创建 vim config/batch.conf

#spark的submit参数，可手动配置
spark {spark.app.name = "Waterdrop"
spark.executor.instances = 2
spark.executor.cores = 2
spark.executor.memory = "2g"
}
#hive的查询语句（table_name为spark的临时表，名字随意）
input {hive {pre_sql = "select * from access.nginx_msg_detail"
table_name = "access_log"
}
}
#可以过滤掉input中不需要写入clickhouse的字段（无过滤可不填写）
filter {remove {source_field = ["minute", "hour"]
}
}
#clickhouse的参数配置
output {clickhouse {host = "your.clickhouse.host:8123"
database = "waterdrop"
table = "access_log" #clickHouse中的表名，需提前建好
fields = ["date", "datetime", "hostname", "uri", "http_code", "request_time", "data_size", "domain"] #写入clickhouse的字段
username = "username"
password = "password"
}
}

output还有部分参数选配：
clickhouse.socket_timeout = 60000 --超时时间

bulk_size = 20000 --批次大小，默认2万条，可适当加大

retry_codes = [209, 210]

retry = 3 --重试次数

执行命令，指定配置文件，运行Waterdrop，即可将数据写入ClickHouse。这里我们以本地模式为例。

./bin/start-waterdrop.sh --config config/batch.conf -e client -m 'local[2]'

在Yarn集群上运行Waterdrop

client 模式 ： ./bin/start-waterdrop.sh --master yarn --queue xxxx  --deploy-mode client --config ./config/application.conf
cluster 模式 ：./bin/start-waterdrop.sh --master yarn --queue xxxx  --deploy-mode cluster --config ./config/application.conf