优秀的数据工程师，怎么用 Spark 在 TiDB 上做 OLAP 分析

作者：RickyHuo本文转载自公众号「大道至简bigdata」原文链接：优秀的数据工程师，怎么用 Spark 在 TiDB 上做 OLAP 分析TiDB 是一款定位于在线事务处理/在线分析处理的融合型数据库产品，实现了一键水平伸缩，强一致性的多副本数据安全，分布式事务，实时 OLAP 等重要特性。TiSpark 是 PingCAP 为解决用户复杂 OLAP 需求而推出的产品。它借助 Spark 平台，同时融合 TiKV 分布式集群的优势。直接使用 TiSpark 完成 OLAP 操作需要了解 Spark，还需要一些开发工作。那么，有没有一些开箱即用的工具能帮我们更快速地使用 TiSpark 在 TiDB 上完成 OLAP 分析呢？目前开源社区上有一款工具 Waterdrop，可以基于 Spark，在 TiSpark 的基础上快速实现 TiDB 数据读取和 OLAP 分析。项目地址：https://github.com/InterestingLab/waterdrop使用 Waterdrop 操作 TiDB在我们线上有这么一个需求，从 TiDB 中读取某一天的网站访问数据，统计每个域名以及服务返回状态码的访问次数，最后将统计结果写入 TiDB 另外一个表中。 我们来看看 Waterdrop 是如何实现这么一个功能的。WaterdropWaterdrop 是一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在 Spark 之上。Waterdrop 拥有着非常丰富的插件，支持从 TiDB、Kafka、HDFS、Kudu 中读取数据，进行各种各样的数据处理，然后将结果写入 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（存储结果数据的表）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，安装十分简单，无需配置系统环境变量1) 准备 Spark 环境2) 安装 Waterdrop3) 配置 Waterdrop以下是简易步骤，具体安装可以参照 Quick Start。# 下载安装Sparkcd /usr/localwget https://archive.apache.org/dist/spark/spark-2.1.0/spark-2.1.0-bin-hadoop2.7.tgztar -xvf https://archive.apache.org/dist/spark/spark-2.1.0/spark-2.1.0-bin-hadoop2.7.tgzwget# 下载安装Waterdrophttps://github.com/InterestingLab/waterdrop/releases/download/v1.2.0/waterdrop-1.2.0.zipunzip waterdrop-1.2.0.zipcd waterdrop-1.2.0vim config/waterdrop-env.sh# 指定Spark安装路径SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.1.0-bin-hadoop2.7}实现 Waterdrop 处理流程我们仅需要编写一个 Waterdrop 配置文件即可完成数据的读取、处理、写入。Waterdrop 配置文件由四个部分组成，分别是 Spark、Input、Filter 和 Output。Input 部分用于指定数据的输入源，Filter 部分用于定义各种各样的数据处理、聚合，Output 部分负责将处理之后的数据写入指定的数据库或者消息队列。整个处理流程为 Input -> Filter -> Output，整个流程组成了 Waterdrop 的处理流程（Pipeline）。以下是一个具体配置，此配置来源于线上实际应用，但是为了演示有所简化。Input (TiDB)这里部分配置定义输入源，如下是从 TiDB 一张表中读取数据。input { tidb { database = “nginx” pre_sql = “select * from nginx.access_log” table_name = “spark_nginx_input” }}Filter在 Filter 部分，这里我们配置一系列的转化, 大部分数据分析的需求，都是在 Filter 完成的。Waterdrop 提供了丰富的插件，足以满足各种数据分析需求。这里我们通过 SQL 插件完成数据的聚合操作。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 实现的。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 执行时所需的资源大小以及其他 Spark 配置。我们的 TiDB Input 插件是基于 TiSpark 实现的，而 TiSpark 依赖于 TiKV 集群和 Placement Driver (PD)。因此我们需要指定 PD 节点信息以及 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.confspark { 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” }}执行命令，指定配置文件，运行 Waterdrop ，即可实现我们的数据处理逻辑。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）就可以了，一般生产环境下，都是使用 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 中读取数据，做简单的数据处理之后写入 TiDB 另外一个表中。仅通过一个配置文件便可快速完成数据的导入，无需编写任何代码。除了支持 TiDB 数据源之外，Waterdrop 同样支持 Elasticsearch，Kafka，Kudu， ClickHouse 等数据源。与此同时，我们正在研发一个重要功能，就是在 Waterdrop 中，利用 TiDB 的事务特性，实现从 Kafka 到 TiDB 流式数据处理，并且支持端（Kafka）到端（TiDB）的 Exactly-Once 数据一致性。希望了解 Waterdrop 和 TiDB，ClickHouse、Elasticsearch、Kafka 结合使用的更多功能和案例，可以直接进入项目主页：https://github.com/InterestingLab/waterdrop ，或者联系项目负责人： Garyelephan（微信: garyelephant）、RickyHuo （微信: chodomatte1994）。