如何快速地把HDFS中的数据导入ClickHouse

霍霍霍霍晨 · · 1748 次点击 · · 开始浏览

这是一个创建于的文章，其中的信息可能已经有所发展或是发生改变。

如何快速地把HDFS中的数据导入ClickHouse

3739fb5b1bad134c6fc2ebccb502178d.jpeg

ClickHouse是面向OLAP的分布式列式DBMS。我们部门目前已经把所有数据分析相关的日志数据存储至ClickHouse这个优秀的数据仓库之中，当前日数据量达到了300亿。

之前介绍的有关数据处理入库的经验都是基于实时数据流，数据存储在Kafka中，我们使用Java或者Golang将数据从Kafka中读取、解析、清洗之后写入ClickHouse中，这样可以实现数据的快速接入。然而在很多同学的使用场景中，数据都不是实时的，可能需要将HDFS或者是Hive中的数据导入ClickHouse。有的同学通过编写Spark程序来实现数据的导入，那么是否有更简单、高效的方法呢。

目前开源社区上有一款工具Waterdrop，项目地址https://github.com/InterestingLab/waterdrop，可以快速地将HDFS中的数据导入ClickHouse。

HDFS to ClickHouse

假设我们的日志存储在HDFS中，我们需要将日志进行解析并筛选出我们关心的字段，将对应的字段写入ClickHouse的表中。

Log Sample

我们在HDFS中存储的日志格式如下，是很常见的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建表语句如下，我们的表按日进行分区

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

接下来会给大家详细介绍，我们如何通过Waterdrop满足上述需求，将HDFS中的数据写入ClickHouse中。

Waterdrop

Waterdrop是一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在Spark之上。Waterdrop拥有着非常丰富的插件，支持从Kafka、HDFS、Kudu中读取数据，进行各种各样的数据处理，并将结果写入ClickHouse、Elasticsearch或者Kafka中。

Prerequisites

首先我们需要安装Waterdrop，安装十分简单，无需配置系统环境变量

准备Spark环境
安装Waterdrop
配置Waterdrop

以下是简易步骤，具体安装可以参照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

我们仅需要编写一个Waterdrop Pipeline的配置文件即可完成数据的导入。

配置文件包括四个部分，分别是Spark、Input、filter和Output。

Spark

这一部分是Spark的相关配置，主要配置Spark执行时所需的资源大小。

spark {
  spark.app.name = "Waterdrop"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
}

Input

这一部分定义数据源，如下是从HDFS文件中读取text格式数据的配置案例。

input {
    hdfs {
        path = "hdfs://nomanode:8020/rowlog/accesslog"
        table_name = "access_log"
        format = "text"
    }
}

Filter

在Filter部分，这里我们配置一系列的转化，包括正则解析将日志进行拆分、时间转换将HTTPDATE转化为ClickHouse支持的日期格式、对Number类型的字段进行类型转换以及通过SQL进行字段筛减等

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"格式的数据转换为
    # "yyyy/MM/dd HH:mm:ss"格式的数据
    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筛选关注的字段，并对字段进行处理
    # 甚至可以通过过滤条件过滤掉不关心的数据
    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

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"格式的数据转换为
    # "yyyy/MM/dd HH:mm:ss"格式的数据
    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筛选关注的字段，并对字段进行处理
    # 甚至可以通过过滤条件过滤掉不关心的数据
    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"
    }
}

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

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

Conclusion

在这篇文章中，我们介绍了如何使用Waterdrop将HDFS中的Nginx日志文件导入ClickHouse中。仅通过一个配置文件便可快速完成数据的导入，无需编写任何代码。除了支持HDFS数据源之外，Waterdrop同样支持将数据从Kafka中实时读取处理写入ClickHouse中。我们的下一篇文章将会介绍，如何将Hive中的数据快速导入ClickHouse中。

当然，Waterdrop不仅仅是ClickHouse数据写入的工具，在Elasticsearch以及Kafka等数据源的写入上同样可以扮演相当重要的角色。

希望了解Waterdrop和ClickHouse、Elasticsearch、Kafka结合使用的更多功能和案例，可以直接进入项目主页https://github.com/InterestingLab/waterdrop

-- Power by InterestingLab

有疑问加站长微信联系（非本文作者）

本文来自：简书

感谢作者：霍霍霍霍晨

查看原文：如何快速地把HDFS中的数据导入ClickHouse

入群交流（和以上内容无关）：加入Go大咖交流群，或添加微信：liuxiaoyan-s 备注：入群；或加QQ群：692541889

1748 次点击

加入收藏微博

收入我的专栏

上一篇：Hyperledger Fabric智能合约

下一篇：golang简易工作池

字段

github

配置文件

vim

0 回复

添加一条新回复（您需要登录后才能回复没有账号？）

请尽量让自己的回复能够对别人有帮助
支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
支持 @ 本站用户；支持表情（输入 : 提示），见 Emoji cheat sheet
图片支持拖拽、截图粘贴等方式上传

关注我

扫码关注领全套学习资料
加入 QQ 群：
- 192706294（已满）
- 731990104（已满）
- 798786647（已满）
- 729884609（已满）
- 977810755（已满）
- 815126783（已满）
- 812540095（已满）
- 1006366459（已满）
- 692541889
加入微信群：liuxiaoyan-s，备注入群
也欢迎加入知识星球 Go粉丝们（免费）

如何快速地把HDFS中的数据导入ClickHouse

如何快速地把HDFS中的数据导入ClickHouse

HDFS to ClickHouse

Log Sample

ClickHouse Schema

Waterdrop with ClickHouse

Waterdrop

Prerequisites

Waterdrop Pipeline

Spark

Input

Filter

Output

Running Waterdrop

Conclusion

用户登录

今日阅读排行

一周阅读排行

关注我

如何快速地把HDFS中的数据导入ClickHouse

HDFS to ClickHouse

Log Sample

ClickHouse Schema

Waterdrop with ClickHouse

Waterdrop

Prerequisites

Waterdrop Pipeline

Spark

Input

Filter

Output

Running Waterdrop

Conclusion

如何快速地把HDFS中的数据导入ClickHouse

如何快速地把HDFS中的数据导入ClickHouse

HDFS to ClickHouse

Log Sample

ClickHouse Schema

Waterdrop with ClickHouse

Waterdrop

Prerequisites

Waterdrop Pipeline

Spark

Input

Filter

Output

Running Waterdrop

Conclusion

用户登录

今日阅读排行

一周阅读排行

关注我

给该专栏投稿 写篇新文章

收入到我管理的专栏 新建专栏

如何快速地把HDFS中的数据导入ClickHouse

HDFS to ClickHouse

Log Sample

ClickHouse Schema

Waterdrop with ClickHouse

Waterdrop

Prerequisites

Waterdrop Pipeline

Spark

Input

Filter

Output

Running Waterdrop

Conclusion

给该专栏投稿写篇新文章

收入到我管理的专栏新建专栏