spark 写tidb_优秀的数据工程师，怎么用Spark在TiDB上做OLAP分析
TiDB 是一款定位于在線事務(wù)處理/在線分析處理的融合型數(shù)據(jù)庫(kù)產(chǎn)品，實(shí)現(xiàn)了一鍵水平伸縮，強(qiáng)一致性的多副本數(shù)據(jù)安全，分布式事務(wù)，實(shí)時(shí) OLAP 等重要特性。
TiSpark 是 PingCAP 為解決用戶復(fù)雜 OLAP 需求而推出的產(chǎn)品。它借助 Spark 平臺(tái)，同時(shí)融合 TiKV 分布式集群的優(yōu)勢(shì)。
直接使用 TiSpark 完成 OLAP 操作需要了解 Spark，還需要一些開發(fā)工作。那么，有沒有一些開箱即用的工具能幫我們更快速地使用 TiSpark 在 TiDB 上完成 OLAP 分析呢？
目前開源社區(qū)上有一款工具 Waterdrop，項(xiàng)目地址 https://github.com/InterestingLab/waterdrop ，可以基于Spark，在 TiSpark 的基礎(chǔ)上快速實(shí)現(xiàn) TiDB 數(shù)據(jù)讀取和 OLAP 分析。
使用Waterdrop操作TiDB
在我們線上有這么一個(gè)需求，從 TiDB 中讀取某一天的網(wǎng)站訪問數(shù)據(jù)，統(tǒng)計(jì)每個(gè)域名以及服務(wù)返回狀態(tài)碼的訪問次數(shù)，最后將統(tǒng)計(jì)結(jié)果寫入 TiDB 另外一個(gè)表中。 我們來(lái)看看Waterdrop是如何實(shí)現(xiàn)這么一個(gè)功能的。
Waterdrop
Waterdrop 是一個(gè)非常易用，高性能，能夠應(yīng)對(duì)海量數(shù)據(jù)的實(shí)時(shí)數(shù)據(jù)處理產(chǎn)品，它構(gòu)建在 Spark 之上。Waterdrop 擁有著非常豐富的插件，支持從 TiDB、Kafka、HDFS、Kudu 中讀取數(shù)據(jù)，進(jìn)行各種各樣的數(shù)據(jù)處理，然后將結(jié)果寫入 TiDB、ClickHouse、Elasticsearch 或者 Kafka 中。
準(zhǔn)備工作
1. TiDB 表結(jié)構(gòu)介紹
Input(存儲(chǔ)訪問日志的表)
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(存儲(chǔ)結(jié)果數(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，安裝十分簡(jiǎn)單，無(wú)需配置系統(tǒng)環(huán)境變量 1. 準(zhǔn)備 Spark環(huán)境 2. 安裝 Waterdrop 3. 配置 Waterdrop
以下是簡(jiǎn)易步驟，具體安裝可以參照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}
實(shí)現(xiàn) Waterdrop 處理流程
我們僅需要編寫一個(gè) Waterdrop 配置文件即可完成數(shù)據(jù)的讀取、處理、寫入。
Waterdrop 配置文件由四個(gè)部分組成，分別是 Spark、Input、Filter 和 Output。Input 部分用于指定數(shù)據(jù)的輸入源，Filter 部分用于定義各種各樣的數(shù)據(jù)處理、聚合，Output 部分負(fù)責(zé)將處理之后的數(shù)據(jù)寫入指定的數(shù)據(jù)庫(kù)或者消息隊(duì)列。
整個(gè)處理流程為 Input -> Filter -> Output，整個(gè)流程組成了 Waterdrop 的 處理流程(Pipeline)。以下是一個(gè)具體配置，此配置來(lái)源于線上實(shí)際應(yīng)用，但是為了演示有所簡(jiǎn)化。
Input (TiDB)
這里部分配置定義輸入源，如下是從 TiDB 一張表中讀取數(shù)據(jù)。
input {
tidb {
database = "nginx"
pre_sql = "select * from nginx.access_log"
table_name = "spark_nginx_input"
}
}
Filter
在Filter部分，這里我們配置一系列的轉(zhuǎn)化, 大部分?jǐn)?shù)據(jù)分析的需求，都是在Filter完成的。Waterdrop 提供了豐富的插件，足以滿足各種數(shù)據(jù)分析需求。這里我們通過(guò) 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)
最后， 我們將處理后的結(jié)果寫入TiDB另外一張表中。TiDB Output是通過(guò)JDBC實(shí)現(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 的相關(guān)配置，主要配置 Spark 執(zhí)行時(shí)所需的資源大小以及其他 Spark 配置。
我們的 TiDB Input 插件是基于 TiSpark 實(shí)現(xiàn)的，而 TiSpark 依賴于 TiKV 集群和 Placement Driver (PD)。因此我們需要指定 PD 節(jié)點(diǎn)信息以及 TiSpark 相關(guān)配置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"
}
運(yùn)行 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í)行命令，指定配置文件，運(yùn)行 Waterdrop ，即可實(shí)現(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
如果是本機(jī)測(cè)試驗(yàn)證邏輯，用本地模式(Local)就可以了，一般生產(chǎn)環(huán)境下，都是使用yarn-client或者yarn-cluster模式。
檢查結(jié)果
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)
總結(jié)
在這篇文章中，我們介紹了如何使用 Waterdrop 從 TiDB 中讀取數(shù)據(jù)，做簡(jiǎn)單的數(shù)據(jù)處理之后寫入 TiDB 另外一個(gè)表中。僅通過(guò)一個(gè)配置文件便可快速完成數(shù)據(jù)的導(dǎo)入，無(wú)需編寫任何代碼。
除了支持 TiDB 數(shù)據(jù)源之外，Waterdrop 同樣支持Elasticsearch, Kafka, Kudu, ClickHouse等數(shù)據(jù)源。
于此同時(shí)，我們正在研發(fā)一個(gè)重要功能，就是在 Waterdrop 中，利用 TiDB 的事務(wù)特性，實(shí)現(xiàn)從 Kafka 到 TiDB 流式數(shù)據(jù)處理，并且支持端(Kafka)到端(TiDB)的 Exactly-Once 數(shù)據(jù)一致性。
希望了解 Waterdrop 和 TiDB，ClickHouse、Elasticsearch、Kafka結(jié)合使用的更多功能和案例，可以直接進(jìn)入項(xiàng)目主頁(yè) https://github.com/InterestingLab/waterdrop或者聯(lián)系項(xiàng)目負(fù)責(zé)人：Garyelephan 微信: garyelephant RickyHuo 微信: chodomatte1994
總結(jié)
以上是生活随笔為你收集整理的spark 写tidb_优秀的数据工程师，怎么用Spark在TiDB上做OLAP分析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： vgg16卷积层的计算量_深度学习_经典
- 下一篇： mupdf不支持x64_x86平台转x6