内容简介:ClickHouse是面向OLAP的分布式列式DBMS。我们部门目前已经把所有数据分析相关的日志数据存储至ClickHouse这个优秀的数据仓库之中,当前日数据量达到了300亿。之前介绍的有关数据处理入库的经验都是基于实时数据流,数据存储在Kafka中,我们使用Java或者Golang将数据从Kafka中读取、解析、清洗之后写入ClickHouse中,这样可以实现数据的快速接入。然而在很多同学的使用场景中,数据都不是实时的,可能需要将HDFS或者是Hive中的数据导入ClickHouse。有的同学通过编写
如何快速地把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
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网
猜你喜欢:- Phoenix 数据导入与导出
- Redis批量导入数据的方法
- 导入Blob数据到Hive
- MySQL也能并发导入数据
- JanusGraph批量导入数据代码总结
- 数据搬运组件:基于 Sqoop 管理数据导入和导出
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
计算机程序设计艺术(第1卷)
[美] 唐纳德·E. 克努特 / 苏运霖 / 国防工业出版社 / 2002-9 / 98.00元
7卷本《计算机程序设计艺术》的第1卷以基本的程序设计概念和技术开始,然后专注于信息结构——计算机内部信息的表示、数据元素之间的结构关系以及如何有效地处理它们,给出了对于模拟、数值方法、符号计算、软件和系统设计的初等应用。书中附有大量习题和答案,标明了难易程序及数学概念的使用。 此新版本增加了几十项简单且重要的算法和技术,并对有关数学预备知识作了大量修改以适应现时研究的趋势。一起来看看 《计算机程序设计艺术(第1卷)》 这本书的介绍吧!