TiDB 是一款定位于在線事務處理/在線分析處理的融合型數(shù)據(jù)庫產品，實現(xiàn)了一鍵水平伸縮，強一致性的多副本數(shù)據(jù)安全，分布式事務，實時 OLAP 等重要特性。

TiSpark 是 PingCAP 為解決用戶復雜 OLAP 需求而推出的產品。它借助 Spark 平臺，同時融合 TiKV 分布式集群的優(yōu)勢。

直接使用 TiSpark 完成 OLAP 操作需要了解 Spark，還需要一些開發(fā)工作。那么，有沒有一些開箱即用的工具能幫我們更快速地使用 TiSpark 在 TiDB 上完成 OLAP 分析呢？

目前開源社區(qū)上有一款工具 Waterdrop，項目地址 https://github.com/InterestingLab/waterdrop ，可以基于Spark，在 TiSpark 的基礎上快速實現(xiàn) TiDB 數(shù)據(jù)讀取和 OLAP 分析。

使用Waterdrop操作TiDB

在我們線上有這么一個需求，從 TiDB 中讀取某一天的網(wǎng)站訪問數(shù)據(jù)，統(tǒng)計每個域名以及服務返回狀態(tài)碼的訪問次數(shù)，最后將統(tǒng)計結果寫入 TiDB 另外一個表中。 我們來看看Waterdrop是如何實現(xiàn)這么一個功能的。

Waterdrop

Waterdrop 是一個非常易用，高性能，能夠應對海量數(shù)據(jù)的實時數(shù)據(jù)處理產品，它構建在 Spark 之上。Waterdrop 擁有著非常豐富的插件，支持從 TiDB、Kafka、HDFS、Kudu 中讀取數(shù)據(jù)，進行各種各樣的數(shù)據(jù)處理，然后將結果寫入 TiDB、ClickHouse、Elasticsearch 或者 Kafka 中。

準備工作

1. TiDB 表結構介紹

Input(存儲訪問日志的表)

CREATE TABLE access_log (

domain VARCHAR(255),

datetime VARCHAR(63),

remote_addr VARCHAR(63),

http_ver VARCHAR(15),

body_bytes_send INT,

status INT,

request_time FLOAT,

url TEXT

)

+-----------------+--------------+------+------+---------+-------+

| Field | Type | Null | Key | Default | Extra |

+-----------------+--------------+------+------+---------+-------+

| domain | varchar(255) | YES | | NULL | |

| datetime | varchar(63) | YES | | NULL | |

| remote_addr | varchar(63) | YES | | NULL | |

| http_ver | varchar(15) | YES | | NULL | |

| body_bytes_send | int(11) | YES | | NULL | |

| status | int(11) | YES | | NULL | |

| request_time | float | YES | | NULL | |

| url | text | YES | | NULL | |

+-----------------+--------------+------+------+---------+-------+

Output(存儲結果數(shù)據(jù)的表)

CREATE TABLE access_collect (

date VARCHAR(23),

domain VARCHAR(63),

status INT,

hit INT

)

+--------+-------------+------+------+---------+-------+

| Field | Type | Null | Key | Default | Extra |

+--------+-------------+------+------+---------+-------+

| date | varchar(23) | YES | | NULL | |

| domain | varchar(63) | YES | | NULL | |

| status | int(11) | YES | | NULL | |

| hit | int(11) | YES | | NULL | |

+--------+-------------+------+------+---------+-------+

2. 安裝 Waterdrop

有了 TiDB 輸入和輸出表之后， 我們需要安裝 Waterdrop，安裝十分簡單，無需配置系統(tǒng)環(huán)境變量 1. 準備 Spark環(huán)境 2. 安裝 Waterdrop 3. 配置 Waterdrop

以下是簡易步驟，具體安裝可以參照Quick Start

# 下載安裝Spark

cd /usr/local

wget https://archive.apache.org/dist/spark/spark-2.1.0/spark-2.1.0-bin-hadoop2.7.tgz

tar -xvf https://archive.apache.org/dist/spark/spark-2.1.0/spark-2.1.0-bin-hadoop2.7.tgz

wget

# 下載安裝Waterdrop

https://github.com/InterestingLab/waterdrop/releases/download/v1.2.0/waterdrop-1.2.0.zip

unzip waterdrop-1.2.0.zip

cd waterdrop-1.2.0

vim config/waterdrop-env.sh

# 指定Spark安裝路徑

SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.1.0-bin-hadoop2.7}

實現(xiàn) Waterdrop 處理流程

我們僅需要編寫一個 Waterdrop 配置文件即可完成數(shù)據(jù)的讀取、處理、寫入。

Waterdrop 配置文件由四個部分組成，分別是 Spark、Input、Filter 和 Output。Input 部分用于指定數(shù)據(jù)的輸入源，Filter 部分用于定義各種各樣的數(shù)據(jù)處理、聚合，Output 部分負責將處理之后的數(shù)據(jù)寫入指定的數(shù)據(jù)庫或者消息隊列。

整個處理流程為 Input -> Filter -> Output，整個流程組成了 Waterdrop 的 處理流程(Pipeline)。以下是一個具體配置，此配置來源于線上實際應用，但是為了演示有所簡化。

Input (TiDB)

這里部分配置定義輸入源，如下是從 TiDB 一張表中讀取數(shù)據(jù)。

input {

tidb {

database = "nginx"

pre_sql = "select * from nginx.access_log"

table_name = "spark_nginx_input"

}

}

Filter

在Filter部分，這里我們配置一系列的轉化, 大部分數(shù)據(jù)分析的需求，都是在Filter完成的。Waterdrop 提供了豐富的插件，足以滿足各種數(shù)據(jù)分析需求。這里我們通過 SQL 插件完成數(shù)據(jù)的聚合操作。

filter {

sql {

table_name = "spark_nginx_log"

sql = "select count(*) as hit, domain, status, substring(datetime, 1, 10) as date from spark_nginx_log where substring(datetime, 1, 10)='2019-01-20' group by domain, status, substring(datetime, 1, 10)"

}

}

Output (TiDB)

最后， 我們將處理后的結果寫入TiDB另外一張表中。TiDB Output是通過JDBC實現(xiàn)的

output {

tidb {

url = "jdbc:mysql://127.0.0.1:4000/nginx?useUnicode=true&characterEncoding=utf8"

table = "access_collect"

user = "username"

password = "password"

save_mode = "append"

}

}

Spark

這一部分是 Spark 的相關配置，主要配置 Spark 執(zhí)行時所需的資源大小以及其他 Spark 配置。

我們的 TiDB Input 插件是基于 TiSpark 實現(xiàn)的，而 TiSpark 依賴于 TiKV 集群和 Placement Driver (PD)。因此我們需要指定 PD 節(jié)點信息以及 TiSpark 相關配置spark.tispark.pd.addresses和spark.sql.extensions。

spark {

spark.app.name = "Waterdrop-tidb"

spark.executor.instances = 2

spark.executor.cores = 1

spark.executor.memory = "1g"

# Set for TiSpark

spark.tispark.pd.addresses = "localhost:2379"

spark.sql.extensions = "org.apache.spark.sql.TiExtensions"

}

運行 Waterdrop

我們將上述四部分配置組合成我們最終的配置文件conf/tidb.conf

spark {

spark.app.name = "Waterdrop-tidb"

spark.executor.instances = 2

spark.executor.cores = 1

spark.executor.memory = "1g"

# Set for TiSpark

spark.tispark.pd.addresses = "localhost:2379"

spark.sql.extensions = "org.apache.spark.sql.TiExtensions"

}

input {

tidb {

database = "nginx"

pre_sql = "select * from nginx.access_log"

table_name = "spark_table"

}

}

filter {

sql {

table_name = "spark_nginx_log"

sql = "select count(*) as hit, domain, status, substring(datetime, 1, 10) as date from spark_nginx_log where substring(datetime, 1, 10)='2019-01-20' group by domain, status, substring(datetime, 1, 10)"

}

}

output {

tidb {

url = "jdbc:mysql://127.0.0.1:4000/nginx?useUnicode=true&characterEncoding=utf8"

table = "access_collect"

user = "username"

password = "password"

save_mode = "append"

}

}

執(zhí)行命令，指定配置文件，運行 Waterdrop ，即可實現(xiàn)我們的數(shù)據(jù)處理邏輯。Local./bin/start-waterdrop.sh --config config/tidb.conf --deploy-mode client --master 'local[2]'yarn-client./bin/start-waterdrop.sh --config config/tidb.conf --deploy-mode client --master yarnyarn-cluster./bin/start-waterdrop.sh --config config/tidb.conf --deploy-mode cluster -master yarn

如果是本機測試驗證邏輯，用本地模式(Local)就可以了，一般生產環(huán)境下，都是使用yarn-client或者yarn-cluster模式。

檢查結果

mysql> select * from access_collect;

+------------+--------+--------+------+

| date | domain | status | hit |

+------------+--------+--------+------+

| 2019-01-20 | b.com | 200 | 63 |

| 2019-01-20 | a.com | 200 | 85 |

+------------+--------+--------+------+

2 rows in set (0.21 sec)

總結

在這篇文章中，我們介紹了如何使用 Waterdrop 從 TiDB 中讀取數(shù)據(jù)，做簡單的數(shù)據(jù)處理之后寫入 TiDB 另外一個表中。僅通過一個配置文件便可快速完成數(shù)據(jù)的導入，無需編寫任何代碼。

除了支持 TiDB 數(shù)據(jù)源之外，Waterdrop 同樣支持Elasticsearch, Kafka, Kudu, ClickHouse等數(shù)據(jù)源。

于此同時，我們正在研發(fā)一個重要功能，就是在 Waterdrop 中，利用 TiDB 的事務特性，實現(xiàn)從 Kafka 到 TiDB 流式數(shù)據(jù)處理，并且支持端(Kafka)到端(TiDB)的 Exactly-Once 數(shù)據(jù)一致性。

希望了解 Waterdrop 和 TiDB，ClickHouse、Elasticsearch、Kafka結合使用的更多功能和案例，可以直接進入項目主頁 https://github.com/InterestingLab/waterdrop或者聯(lián)系項目負責人：Garyelephan 微信: garyelephant RickyHuo 微信: chodomatte1994

總結

以上是生活随笔為你收集整理的spark 写tidb_优秀的数据工程师，怎么用Spark在TiDB上做OLAP分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內容還不錯，歡迎將生活随笔推薦給好友。