如何快速地把HDFS中的數(shù)據(jù)導(dǎo)入ClickHouse
ClickHouse是面向OLAP的分布式列式DBMS鸡典。我們部門目前已經(jīng)把所有數(shù)據(jù)分析相關(guān)的日志數(shù)據(jù)存儲(chǔ)至ClickHouse這個(gè)優(yōu)秀的數(shù)據(jù)倉(cāng)庫(kù)之中，當(dāng)前日數(shù)據(jù)量達(dá)到了300億陪毡。
之前介紹的有關(guān)數(shù)據(jù)處理入庫(kù)的經(jīng)驗(yàn)都是基于實(shí)時(shí)數(shù)據(jù)流讹开，數(shù)據(jù)存儲(chǔ)在Kafka中侠姑，我們使用Java或者Golang將數(shù)據(jù)從Kafka中讀取篷就、解析汹粤、清洗之后寫入ClickHouse中命斧，這樣可以實(shí)現(xiàn)數(shù)據(jù)的快速接入。然而在很多同學(xué)的使用場(chǎng)景中嘱兼，數(shù)據(jù)都不是實(shí)時(shí)的国葬，可能需要將HDFS或者是Hive中的數(shù)據(jù)導(dǎo)入ClickHouse。有的同學(xué)通過編寫Spark程序來實(shí)現(xiàn)數(shù)據(jù)的導(dǎo)入芹壕，那么是否有更簡(jiǎn)單胃惜、高效的方法呢。
目前開源社區(qū)上有一款工具Waterdrop哪雕，項(xiàng)目地址https://github.com/InterestingLab/waterdrop，可以快速地將HDFS中的數(shù)據(jù)導(dǎo)入ClickHouse鲫趁。
HDFS to ClickHouse
假設(shè)我們的日志存儲(chǔ)在HDFS中斯嚎，我們需要將日志進(jìn)行解析并篩選出我們關(guān)心的字段，將對(duì)應(yīng)的字段寫入ClickHouse的表中。
Log Sample
我們?cè)贖DFS中存儲(chǔ)的日志格式如下堡僻， 是很常見的Nginx日志
10.41.1.28 github.com 114.250.140.241 0.001s "127.0.0.1:80" [26/Oct/2018:03:09:32 +0800] "GET /InterestingLab/waterdrop HTTP/1.1" 200 0 "-" - "Dalvik/2.1.0 (Linux; U; Android 7.1.1; OPPO R11 Build/NMF26X)" "196" "-" "mainpage" "443" "-" "172.16.181.129"
ClickHouse Schema
我們的ClickHouse建表語(yǔ)句如下糠惫，我們的表按日進(jìn)行分區(qū)
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,
pool String
) ENGINE = MergeTree PARTITION BY date ORDER BY date SETTINGS index_granularity = 16384
Waterdrop with ClickHouse
接下來會(huì)給大家詳細(xì)介紹，我們?nèi)绾瓮ㄟ^Waterdrop滿足上述需求钉疫，將HDFS中的數(shù)據(jù)寫入ClickHouse中硼讽。
Waterdrop
Waterdrop是一個(gè)非常易用，高性能牲阁，能夠應(yīng)對(duì)海量數(shù)據(jù)的實(shí)時(shí)數(shù)據(jù)處理產(chǎn)品固阁，它構(gòu)建在Spark之上。Waterdrop擁有著非常豐富的插件城菊，支持從Kafka备燃、HDFS、Kudu中讀取數(shù)據(jù)凌唬，進(jìn)行各種各樣的數(shù)據(jù)處理并齐，并將結(jié)果寫入ClickHouse、Elasticsearch或者Kafka中客税。
Prerequisites
首先我們需要安裝Waterdrop况褪，安裝十分簡(jiǎn)單，無需配置系統(tǒng)環(huán)境變量
- 準(zhǔn)備Spark環(huán)境
- 安裝Waterdrop
- 配置Waterdrop
以下是簡(jiǎn)易步驟更耻，具體安裝可以參照Quick Start
cd /usr/local
wget https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
tar -xvf https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
wget https://github.com/InterestingLab/waterdrop/releases/download/v1.1.1/waterdrop-1.1.1.zip
unzip waterdrop-1.1.1.zip
cd waterdrop-1.1.1
vim config/waterdrop-env.sh
# 指定Spark安裝路徑
SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.2.0-bin-hadoop2.7}
Waterdrop Pipeline
我們僅需要編寫一個(gè)Waterdrop Pipeline的配置文件即可完成數(shù)據(jù)的導(dǎo)入测垛。
配置文件包括四個(gè)部分，分別是Spark酥夭、Input赐纱、filter和Output。
Spark
這一部分是Spark的相關(guān)配置熬北，主要配置Spark執(zhí)行時(shí)所需的資源大小疙描。
spark {
spark.app.name = "Waterdrop"
spark.executor.instances = 2
spark.executor.cores = 1
spark.executor.memory = "1g"
}
Input
這一部分定義數(shù)據(jù)源，如下是從HDFS文件中讀取text格式數(shù)據(jù)的配置案例讶隐。
input {
hdfs {
path = "hdfs://nomanode:8020/rowlog/accesslog"
table_name = "access_log"
format = "text"
}
}
Filter
在Filter部分起胰，這里我們配置一系列的轉(zhuǎn)化，包括正則解析將日志進(jìn)行拆分巫延、時(shí)間轉(zhuǎn)換將HTTPDATE轉(zhuǎn)化為ClickHouse支持的日期格式效五、對(duì)Number類型的字段進(jìn)行類型轉(zhuǎn)換以及通過SQL進(jìn)行字段篩減等
filter {
# 使用正則解析原始日志
grok {
source_field = "raw_message"
pattern = '%{IP:ha_ip}\\s%{NOTSPACE:domain}\\s%{IP:remote_addr}\\s%{NUMBER:request_time}s\\s\"%{DATA:upstream_ip}\"\\s\\[%{HTTPDATE:timestamp}\\]\\s\"%{NOTSPACE:method}\\s%{DATA:url}\\s%{NOTSPACE:http_ver}\"\\s%{NUMBER:status}\\s%{NUMBER:body_bytes_send}\\s%{DATA:referer}\\s%{NOTSPACE:cookie_info}\\s\"%{DATA:user_agent}\"\\s%{DATA:uid}\\s%{DATA:session_id}\\s\"%{DATA:pool}\"\\s\"%{DATA:tag2}\"\\s%{DATA:tag3}\\s%{DATA:tag4}'
}
# 將"dd/MMM/yyyy:HH:mm:ss Z"格式的數(shù)據(jù)轉(zhuǎn)換為
# "yyyy/MM/dd HH:mm:ss"格式的數(shù)據(jù)
date {
source_field = "timestamp"
target_field = "datetime"
source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
target_time_format = "yyyy/MM/dd HH:mm:ss"
}
# 使用SQL篩選關(guān)注的字段，并對(duì)字段進(jìn)行處理
# 甚至可以通過過濾條件過濾掉不關(guān)心的數(shù)據(jù)
sql {
table_name = "access"
sql = "select substring(date, 1, 10) as date, datetime, hostname, url, http_code, float(request_time), int(data_size), domain from access"
}
}
Output
最后我們將處理好的結(jié)構(gòu)化數(shù)據(jù)寫入ClickHouse
output {
clickhouse {
host = "your.clickhouse.host:8123"
database = "waterdrop"
table = "access_log"
fields = ["date", "datetime", "hostname", "uri", "http_code", "request_time", "data_size", "domain"]
username = "username"
password = "password"
}
}
Running Waterdrop
我們將上述四部分配置組合成為我們的配置文件config/batch.conf炉峰。
vim config/batch.conf
spark {
spark.app.name = "Waterdrop"
spark.executor.instances = 2
spark.executor.cores = 1
spark.executor.memory = "1g"
}
input {
hdfs {
path = "hdfs://nomanode:8020/rowlog/accesslog"
table_name = "access_log"
format = "text"
}
}
filter {
# 使用正則解析原始日志
grok {
source_field = "raw_message"
pattern = '%{IP:ha_ip}\\s%{NOTSPACE:domain}\\s%{IP:remote_addr}\\s%{NUMBER:request_time}s\\s\"%{DATA:upstream_ip}\"\\s\\[%{HTTPDATE:timestamp}\\]\\s\"%{NOTSPACE:method}\\s%{DATA:url}\\s%{NOTSPACE:http_ver}\"\\s%{NUMBER:status}\\s%{NUMBER:body_bytes_send}\\s%{DATA:referer}\\s%{NOTSPACE:cookie_info}\\s\"%{DATA:user_agent}\"\\s%{DATA:uid}\\s%{DATA:session_id}\\s\"%{DATA:pool}\"\\s\"%{DATA:tag2}\"\\s%{DATA:tag3}\\s%{DATA:tag4}'
}
# 將"dd/MMM/yyyy:HH:mm:ss Z"格式的數(shù)據(jù)轉(zhuǎn)換為
# "yyyy/MM/dd HH:mm:ss"格式的數(shù)據(jù)
date {
source_field = "timestamp"
target_field = "datetime"
source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
target_time_format = "yyyy/MM/dd HH:mm:ss"
}
# 使用SQL篩選關(guān)注的字段畏妖，并對(duì)字段進(jìn)行處理
# 甚至可以通過過濾條件過濾掉不關(guān)心的數(shù)據(jù)
sql {
table_name = "access"
sql = "select substring(date, 1, 10) as date, datetime, hostname, url, http_code, float(request_time), int(data_size), domain from access"
}
}
output {
clickhouse {
host = "your.clickhouse.host:8123"
database = "waterdrop"
table = "access_log"
fields = ["date", "datetime", "hostname", "uri", "http_code", "request_time", "data_size", "domain"]
username = "username"
password = "password"
}
}
執(zhí)行命令，指定配置文件疼阔，運(yùn)行Waterdrop戒劫，即可將數(shù)據(jù)寫入ClickHouse半夷。這里我們以本地模式為例。
./bin/start-waterdrop.sh --config config/batch.conf -e client -m 'local[2]'
Conclusion
在這篇文章中迅细，我們介紹了如何使用Waterdrop將HDFS中的Nginx日志文件導(dǎo)入ClickHouse中巫橄。僅通過一個(gè)配置文件便可快速完成數(shù)據(jù)的導(dǎo)入，無需編寫任何代碼茵典。除了支持HDFS數(shù)據(jù)源之外湘换，Waterdrop同樣支持將數(shù)據(jù)從Kafka中實(shí)時(shí)讀取處理寫入ClickHouse中。我們的下一篇文章將會(huì)介紹统阿，如何將Hive中的數(shù)據(jù)快速導(dǎo)入ClickHouse中彩倚。
當(dāng)然，Waterdrop不僅僅是ClickHouse數(shù)據(jù)寫入的工具砂吞，在Elasticsearch以及Kafka等數(shù)據(jù)源的寫入上同樣可以扮演相當(dāng)重要的角色署恍。
希望了解Waterdrop和ClickHouse、Elasticsearch蜻直、Kafka結(jié)合使用的更多功能和案例盯质，可以直接進(jìn)入項(xiàng)目主頁(yè)https://github.com/InterestingLab/waterdrop
-- Power by InterestingLab
