2. 程式人生 > >如何快速地把HDFS中的資料匯入ClickHouse

如何快速地把HDFS中的資料匯入ClickHouse

阿新 • • 發佈：2018-12-14

如何快速地把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，安裝十分簡單，無需配置系統環境變數

  1. 準備Spark環境
  2. 安裝Waterdrop
  3. 配置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

如何快速地把HDFS中的資料匯入ClickHouse ClickHouse是面向OLAP的分散式列式DBMS。我們部門目前已經把所有資料分析相關的日誌資料儲存至ClickHouse這個優秀的資料倉庫之中，當前日資料量達到了300億。 之前介紹的有關資料處理入庫的經驗都是基於

如何快速地將Hive中的資料匯入ClickHouse

如何快速地將Hive中的資料匯入ClickHouse ClickHouse是面向OLAP的分散式列式DBMS。我們部門目前已經把所有資料分析相關的日誌資料儲存至ClickHouse這個優秀的資料倉庫之中，當前日資料量達到了300億。 在之前的文章如何快速地把HDFS中的資料

MapReduce將HDFS文字資料匯入HBase中

HBase本身提供了很多種資料匯入的方式，通常有兩種常用方式： 使用HBase提供的TableOutputFormat，原理是通過一個Mapreduce作業將資料匯入HBase 另一種方式就是使用HBase原生Client API 本文就是示範如何通過M

【轉載】怎樣把Oracle的資料匯入到SQLServer2005中

怎樣把Oracle的資料匯入到SQLServer2005中 例子：把Oracle的CH330300資料庫的資料匯入到SQLServer2005的Temp資料庫 在本機測試一直沒有成功，一直顯示各種錯誤，網上查詢各種方法無果，下載了ORACLE WIN7 64位客戶端也沒有解決問題，根本就看不到ORACLE的

Java將Excel表格中資料匯入至資料庫中的表中

        上一節介紹了Java將資料庫表中資料匯出至Excel表格，那麼本節來介紹它的逆過程，也就是將Excel表格中的資料逐行匯入資料庫中的表中，依然需要使用Apache的POI，上一節已經說過也附了這個jar包的下載地址，這一節就不過多的說，直接講如

將MySQL中資料匯入到MongoDB中

第一步： 將user表從MySQL中匯出，右鍵，點選匯出嚮導，選擇格式為xlsx。 第二步： 匯出完成後，雙擊開啟user.xlsx，將user.xlsx另存為csv格式的檔案。（切記不可直接修改後綴名，會導致亂碼，無法匯入到MongoDB中，血的教訓） 第三步：

SparkSql將資料來源Hive中資料匯入MySql例項

背景：能看到這篇部落格的夥計兒，應該是充分理解了[理想是豐滿的 現實是骨感]  這句名言了吧。為啥子這麼說呢，那就是 不就是個SparkSql從hive匯入到mysql嗎 有什麼技術含量，但是呢 不斷地踩坑ing填坑ing。 廢話不多說，直接上硬菜。 package co

php 爬蟲的簡單實現， 獲取整個頁面， 再把頁面的資料匯入本地的檔案當中

$curlobj = curl_init(); //建立一個curl 的資源，下面要用的 curl_setopt($curlobj,CURLOPT_URL,"http://www.baidu.com

利用groovy把表中資料匯出成txt或csv

膠水語言就是膠水,寫起也挺快的.這個指令碼主要是從資料庫中把表中的資料匯出來生成檔案.而不用每次都開啟資料庫編輯器去手工收集.然後結合強大的指令碼(shell,bat)等基本就可以實現定時生成最新資料文字了.目前這個gr

C#把txt的資料匯入excel,用逗號分隔

using System; using System.Collections.Generic; using System.ComponentModel; using System.Data; using System.Drawing; using System.Linq; using System.Text;

SpringMVC 實現POI讀取Excle檔案中資料匯入資料庫（上傳）、匯出資料庫中資料到Excle檔案中（下載）

package com.shiliu.game.utils; import java.io.OutputStream; import java.net.URLEncoder; import java.util.ArrayList; import java.util.List; import

把txt原始資料匯入到matlab

第一種方法： importdata max_out.txt; 這種方法匯入的資料儲存在”ans”變數中； 第二種方法： load max_out.txt; 這種方法匯入的資料會儲存在名字為”

怎樣把 excel 的資料匯入到資料庫裡面去

相關文章1. 把 excel 另存為 .csv 格式2. 用 Notepad 開啟 .csv 檔案, 第一行就是所有的欄位3. 建立表結構create table yu_rt_01 as  select Transaction_Id,Last_Update_Date,Last

linux下大檔案編碼轉碼及將oracle中資料匯入mysql

        這篇文章有不少廢話，只是為了發洩一下。如果讀者找需要解決的問題的辦法，直接無視這些廢話。             最近做的專案要將Oracle中資料匯入Mysql， 資料量約有兩千萬條，慶幸的是隻有一張表，而且資料結構比較簡單。在這個過程中遇到不少坑，現在

使用C#將Excel檔案中資料匯入SQL Server資料庫

由於專案中加入了新的功能，可以使管理員向資料庫中匯入Excel資料。因此，在商品管理這塊需要對Excel進行操作，在網上查了些資料，根據專案的實際情況進行了一定的優化，這裡簡單的介紹下。 C#程式碼 /// <summary> /// 上傳Excel檔案，並將資

MongoDB中資料匯入

MongoDB 是一款NoSql資料庫，既然是資料庫，那麼就需要有資料來提供給我們使用，下面我們首先匯入資料。 匯入資料 MongoDB官方提供了示例資料庫檔案，我們就使用官方提供的資料，使用test資料庫和restaurants集合。下面是restaura

如何快速地在java中使用Restful API

什麼是Restful API？REST(英文：Representational State Transfer，簡稱REST)，它是一種軟體架構風格、設計風格，而不是標準，只是提供了一組設計原則和約束條件。它主要用於客戶端和伺服器互動類的軟體。基於這個風格設計的軟體可以更簡潔，

怎麼把sql server資料匯入mysql本地資料庫？

tuimu-dev資料庫是mysql資料庫，下面的sqlserver資料庫是sqlserver的，而我要把sqlserver裡面的AIMSEN_BASE表裡面的accont資料導進來，如下圖：選中tuimu-dev-----然後直接下一步。。。。。。  最後  一個按鈕  開

當你無法發現問題所在時，不要簡單地把程式碼或者資料還原

【場景】 這幾天有一位同事需要做與我一樣的功能，而這個功能是我已經開發好了的，他只需直接把我的程式碼拿去修改就可以。結果，我的程式碼可以正常執行，他的程式碼在開啟服務之後，控制檯就不斷有日誌輸出，開發環境進入了宕機狀態。 他花了很長的時間去尋找問題的來源，但

在ASP.NET中將Excel檔案中資料匯入資料庫並顯示進度條

在ASP.NET中經常會遇到Excel檔案匯入資料庫的問題，遇到資料量比較大的時候，最好顯示進度條。 進度條設計是參考網上某牛人提供的程式碼，利用JS實現的，謝謝這個大牛了（忘記了當時記錄下他的大名了，:(）。具體思路：首先將後臺伺服器上ProgressBar.htm 頁面內