如何讓MySQL查詢速度提升10倍以上-愛可生

在這篇文章中我們將討論如何利用 Apache Spark 來提升 MySQL 的查詢性能。

介紹

在我的前一篇文章 Apache Spark with MySQL 中介紹瞭如何利用 Apache Spark 實現數據分析以及如何對大量存放於文本文件的數據進行轉換和分析。瓦迪姆還做了一個基準測試用來比較 MySQL 和 Spark with Parquet 柱狀格式 (使用空中交通性能數據) 二者的性能。 這個測試非常棒，但如果我們不希望將數據從 MySQL 移到其他的存儲系統中，而是繼續在已有的 MySQL 服務器上執行查詢的話，Apache Spark 一樣可以幫到我們！

開始

在已有的 MySQL 服務器之上使用 Apache Spark （無需將數據導出到 Spark 或者 Hadoop 平臺上），這樣至少可以提升 10 倍的查詢性能。使用多個 MySQL 服務器（複製或者 Percona XtraDB Cluster）可以讓我們在某些查詢上得到額外的性能提升。你也可以使用 Spark 的緩存功能來緩存整個 MySQL 查詢結果表。

思路很簡單：Spark 可以通過 JDBC 讀取 MySQL 上的數據，也可以執行 SQL 查詢，因此我們可以直接連接到 MySQL 並執行查詢。那麼為什麼速度會快呢？對一些需要運行很長時間的查詢（如報表或者BI），由於 Spark 是一個大規模並行系統，因此查詢會非常的快。MySQL 只能為每一個查詢分配一個 CPU 核來處理，而 Spark 可以使用所有集群節點的所有核。在下面的例子中，我們會在 Spark 中執行 MySQL 查詢，這個查詢速度比直接在 MySQL 上執行速度要快 5 到 10 倍。

另外，Spark 可以增加“集群”級別的並行機制，在使用 MySQL 複製或者 Percona XtraDB Cluster 的情況下，Spark 可以把查詢變成一組更小的查詢（有點像使用了分區表時可以在每個分區都執行一個查詢），然後在多個 Percona XtraDB Cluster 節點的多個從服務器上並行的執行這些小查詢。最後它會使用 map/reduce 方式將每個節點返回的結果聚合在一起形成完整的結果。

這篇文章跟我之前文章 “Airlines On-Time Performance” 所使用的數據庫是相同的。瓦迪姆創建了一些腳本可以方便的下載這些數據並上傳到 MySQL 數據庫。腳本的下載地址請看 這裡。同時我們這次使用的是 2016年7月26日發佈的 Apache Spark 2.0。

安裝 Apache Spark

使用獨立模式啟動 Apache Spark 是很簡單的，如下幾步即可：

下載 Apache Spark 2.0 並解壓到某目錄

啟動 master.

啟動 slave (worker) 並連接到 master

啟動應用 (spark-shell 或者 spark-sql).

示例:

為了連接到 Spark ，我們可以使用 spark-shell (Scala)、pyspark (Python) 或者 spark-sql。spark-sql 和 MySQL 命令行類似，因此這是最簡單的選擇（你甚至可以用 show tables 命令）。我同時還需要在交互模式下使用 Scala ，因此我選擇的是 spark-shell 。在下面所有的例子中，我都是在 MySQL 和 Spark 上使用相同的 SQL 查詢，所以其實沒多大的不同。

為了讓 Spark 能用上 MySQL 服務器，我們需要驅動程序 Connector/J for MySQL. 下載這個壓縮文件解壓後拷貝 mysql-connector-java-5.1.39-bin.jar 到 spark 目錄，然後在 conf/spark-defaults.conf 中添加類路徑，如下：

利用 Apache Spark 運行 MySQL 查詢

在這個測試中我們使用的一臺擁有 12 核(老的 Intel(R) Xeon(R) CPU L5639 @ 2.13GHz 處理器) 以及 48G 內存，帶有 SSD 磁盤的物理服務器。 在這臺機器上我安裝了 MySQL 並啟動了 Spark 主節點和從節點。

現在我們可以在 Spark 中運行 MySQL 查詢了。首先，從 Spark 目錄中啟動 Shell （在我這裡是 /usr/local/spark ）：

然後我們將連接到 MySQL 服務器並註冊臨時視圖：

這樣我們就為 Spark 創建了一個“數據源”（換句話說就是相當於 Spark 建立了到 MySQL 的連接）。Spark 表名 “ontime” 對應連接到 MySQL 的ontime.ontime_part 表，現在可以在 Spark 中運行 SQL 了，它們是按順序被一一解析並轉換成 MySQL 查詢的。

“partitionColumn” 在這裡非常重要，它告訴 Spark 並行的執行多個查詢，每個分區分配一個查詢執行。

現在我們可以運行查詢：

<code>val sqlDF = sql("select min(year), max(year) as max_year, Carrier, count(*) as cnt, sum(if(ArrDelayMinutes>30, 1, 0)) as flights_delayed, round(sum(if(ArrDelayMinutes>30, 1, 0))/count(*),2) as rate FROM ontime WHERE DayOfWeek not in (6,7) and OriginState not in ('AK', 'HI', 'PR', 'VI') and DestState not in ('AK', 'HI', 'PR', 'VI') and (origin = 'RDU' or dest = 'RDU') GROUP by carrier HAVING cnt > 100000 and max_year > '1990' ORDER by rate DESC, cnt desc LIMIT  10")
sqlDF.show()/<code>

MySQL 查詢示例

讓我們暫時回到 MySQL 來看看這個查詢例子，我選出瞭如下的查詢語句 （來自我以前的文章）：

這個查詢用來查找出每個航空公司航班延誤的架數。此外該查詢還將很智能的計算準點率，考慮到航班數量（我們不希望小航空公司跟大航空公司比較，同時一些老的關閉的航空公司也不在計算範圍之內）。

我選擇這個查詢主要的原因是，這在 MySQL 很難再優化了，所有的這些 WHERE 條件智能過濾掉約 70% 的記錄行。我做了一個基本的計算：

<code>mysql> select count(*) FROM ontime WHERE DayOfWeek not in (6,7) and OriginState not in ('AK', 'HI', 'PR', 'VI') and DestState not in ('AK', 'HI', 'PR', 'VI');
+-----------+| count(*)  |
+-----------+| 108776741 |
+-----------+mysql> select count(*) FROM ontime;
+-----------+| count(*)  |
+-----------+| 152657276 |
+-----------+mysql> select round((108776741/152657276)*100, 2);
+-------------------------------------+| round((108776741/152657276)*100, 2) |
+-------------------------------------+|                               71.26 |
+-------------------------------------+/<code>

表結構如下：

就算有一個“覆蓋”索引，MySQL 也將掃描約 ~70M-100M 行的數據並創建一個臨時表：

下面是 MySQL 查詢的響應時間：

<code>mysql> select min(yearD), max(yearD) as max_year, Carrier, count(*) as cnt, sum(if(ArrDelayMinutes>30, 1, 0)) as flights_delayed, round(sum(if(ArrDelayMinutes>30, 1, 0))/count(*),2) as rate FROM ontime_part WHERE DayOfWeek not in (6,7) and OriginState not in ('AK', 'HI', 'PR', 'VI') and DestState not in ('AK', 'HI', 'PR', 'VI') GROUP by carrier HAVING cnt > 1000 and max_year > '1990' ORDER by rate DESC, cnt desc LIMIT  10;
足足執行了 19 分鐘，這個結果真的讓人爽不起來。/<code>

SQL in Spark

現在我們希望在 Spark 中運行相同的查詢，讓 Spark 從 MySQL 讀取數據。我們創建了一個“數據源”然後執行如下查詢：

<code>scala> val jdbcDF = spark.read.format("jdbc").options(     |   Map("url" ->  "jdbc:mysql://localhost:3306/ontime?user=root&password=mysql",
     |   "dbtable" -> "ontime.ontime_sm",     |   "fetchSize" -> "10000",
     |   "partitionColumn" -> "yeard", "lowerBound" -> "1988", "upperBound" -> "2015", "numPartitions" -> "48"
     |   )).load()
16/08/02 23:24:12 WARN JDBCRelation: The number of partitions is reduced because the specified number of partitions is less than the difference between upper bound and lower bound. Updated number of partitions: 27; Input number of partitions: 48; Lower bound: 1988; Upper bound: 2015.
dbcDF: org.apache.spark.sql.DataFrame = [id: int, YearD: date ... 19 more fields]
scala> jdbcDF.createOrReplaceTempView("ontime")
scala> val sqlDF = sql("select min(yearD), max(yearD) as max_year, Carrier, count(*) as cnt, sum(if(ArrDelayMinutes>30, 1, 0)) as flights_delayed, round(sum(if(ArrDelayMinutes>30, 1, 0))/count(*),2) as rate FROM ontime WHERE OriginState not in ('AK', 'HI', 'PR', 'VI') and DestState not in ('AK', 'HI', 'PR', 'VI') GROUP by carrier HAVING cnt > 1000 and max_year > '1990' ORDER by rate DESC, cnt desc LIMIT  10")
sqlDF: org.apache.spark.sql.DataFrame = [min(yearD): date, max_year: date ... 4 more fields]
scala> sqlDF.show()
Spark-shell 並不會顯示查詢的執行時間，這個可以從 spark-sql 提供的 Web UI 中獲取到。我在 spark-sql 中重新執行相同的查詢：16/08/04 01:44:27 WARN JDBCRelation: The number of partitions is reduced because the specified number of partitions is less than the difference between upper bound and lower bound. Updated number of partitions: 26; Input number of partitions: 48; Lower bound: 1988; Upper bound: 2014.Time taken: 3.864 seconds/<code>

<code>spark-sql> select min(yearD), max(yearD) as max_year, Carrier, count(*) as cnt, sum(if(ArrDelayMinutes>30, 1, 0)) as flights_delayed, round(sum(if(ArrDelayMinutes>30, 1, 0))/count(*),2) as rate FROM ontime WHERE DayOfWeek not in (6,7) and OriginState not in ('AK', 'HI', 'PR', 'VI') and DestState not in ('AK', 'HI', 'PR', 'VI') GROUP by carrier HAVING cnt > 1000 and max_year > '1990' ORDER by rate DESC, cnt desc LIMIT  10;16/08/04 01:45:13 WARN Utils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.可以看到查詢的時間足足快了 10 倍之多（同一臺機器，只有一臺機器）。但是到底這些查詢是怎麼變成 MySQL 查詢的呢？然後為什麼這樣的查詢會快那麼多。讓我們深入到 MySQL 一探究竟：/<code>

深入 MySQL

Spark:

MySQL:

Spark 並行執行了 26 個查詢，棒極了。由於表本身是分區的，因此每個分區一個查詢，但是卻掃描了整個分區：

<code>mysql> explain partitions SELECT `YearD`,`ArrDelayMinutes`,`Carrier` FROM ontime.ontime_part WHERE (((NOT (DayOfWeek IN (6, 7)))) AND ((NOT (OriginState IN ('AK', 'HI', 'PR', 'VI')))) AND ((NOT (DestState IN ('AK', 'HI', 'PR', 'VI'))))) AND (yearD >= 2001 AND yearD < 2002)G
*************************** 1. row ***************************在這種情況下，服務器有 12 核，24 個超線程，可以非常高效的並行執行 26 個查詢，而分區表可以幫助避免資源爭用的問題（我希望 MySQL 可以並行的掃描分區，但在寫的時候不是這樣）。/<code>

另外一件有趣的事是 Spark 可以“推送”一些條件到 MySQL，但只是在 WHERE 語句範圍內。所有的 GROUP BY/ORDER BY 聚合查詢都是直接在 Spark 中執行。它需要從 MySQL 獲取滿足這些條件的數據，這樣就不會再推送 GROUP BY/ORDER BY 到 MySQL 中。

這也意味著如果不帶 WHERE 條件的查詢（例如 "select count(*) as cnt, carrier from ontime group by carrier order by cnt desc limit 10"）只能是從 MySQL 獲取完整的數據並加載到 Spark 中（等於在 MySQL 做所有 GROUP BY 的操作），這種查詢在 Spark 上運行根據其數據量以及索引的使用情況或慢或快沒準，但其要求更多的資源以及更多的內存佔用。上述的查詢轉成 26 個查詢，每個包含一個 “select carrier from ontime_part where (yearD >= N AND yearD < N)”

將整個查詢推送到 MySQL

如果我們想要避免將所有數據從 MySQL 發送到 Spark，我們可以在查詢之上創建一個臨時表（類似 MySQL 在 select 語句中創建臨時表的方法），可以這樣編寫 Scala 代碼：

請注意：

這裡我們不想用到 “partitionColumn” ，否則我們將看到 26 個類似 MySQL: “SELECT yeard, count(*) FROM (select yeard, count(*) from ontime_part group by yeard) tmp where (yearD >= N AND yearD < N)” 這樣的查詢，顯然這種查詢並沒有經過優化

這並非使用 Spark 的好方法，頂多就是應急一下。這樣做唯一的理由就是能將查詢的結果作為一個額外的查詢。

Spark 的查詢緩存

另外一個方案就是緩存查詢結果（甚至是整張表），然後在 Scala 使用 .filter 實現更快速處理。這種做法需要給 Spark 提供足夠多的內存。好消息是我們可以為 Spark 增加額外的節點來為 Spark 集群獲取更多的內存。

Spark SQL 示例:

spark-sql> select min(yearD), max(yearD) as max_year, Carrier, count(*) as cnt, sum(if(ArrDelayMinutes>30, 1, 0)) as flights_delayed, round(sum(if(ArrDelayMinutes>30, 1,0))/count(*),2) as rate FROM ontime_latest WHERE DayOfWeek not in (6,7) and OriginState not in('AK', 'HI', 'PR', 'VI') and DestState not in ('AK', 'HI', 'PR', 'VI') and (origin='RDU' ordest = 'RDU') GROUP by carrier HAVING cnt > 1000 and max_year > '1990' ORDER by rate DESC, cnt desc LIMIT 10

這裡我們在 Spark 中緩存了分區 p2013 和 p2014 。這將從 MySQL 獲取數據並加載進 Spark 中。一旦在這些緩存的數據上執行查詢，可以想象速度要快得多。

利用 Scala 我們可以緩存查詢的結果，然後使用過濾器去獲取我們所需的信息：

<code>val sqlDF = sql("SELECT flightdate, origin, dest, depdelayminutes, arrdelayminutes, carrier, TailNum, Cancelled, Diverted, Distance from ontime")
sqlDF.cache().show()
scala> sqlDF.filter("flightdate='1988-01-01'").count()
res5: Long = 862/<code>

使用 Spark 和 Percona XtraDB Cluster

Spark 也可以用在集群模式下，支持越來越多的節點，因為只從一個單一的 MySQL 服務器上讀取數據是很大的瓶頸。我們可以使用 MySQL 複製的從服務器或者是 Percona XtraDB Cluster (PXC) 節點來作為 Spark 的數據源。為了測試這個，我們在 AWS 上搞來了三個 Percona XtraDB Cluster 節點的集群 ( m4.2xlarge Ubuntu 實例) 然後在每個節點上啟動 Apache Spark：

Node1 (pxc1): Percona Server + Spark Master + Spark worker node + Spark SQL

Node2 (pxc2): Percona Server + Spark worker node

Node3 (pxc3): Percona Server + Spark worker node

所有的 Spark worker 節點使用相同的內存配置參數：

然後我就可以啟動 spark-sql (一樣需要 connector/J 的 JAR 文件拷貝到所有節點):

當創建表時，我仍然使用 localhost 連接到 MySQL (url “jdbc:mysql://localhost:3306/ontime?user=root&password=xxx”). 由於運行 Spark worker 的節點和運行 Percona Cluster 節點是同一個，因此它將使用的是本地連接。然後運行一個 Spark SQL 會將所有 26 個查詢平均分發到三個 MySQL 節點中。

另外我們可以在獨立的主機上運行 Spark 集群，然後連接到 HAProxy ，這樣可以實現多個 Percona XtraDB Cluster 節點的負載均衡。

查詢性能測試

最後我們來看看在三個 AWS Percona XtraDB Cluster 節點上查詢的響應時間測試結果：

查詢 1:

查詢 2:

select dayofweek, count(*) from ontime_part group by dayofweek;

這個結果看起來很棒，但還可以更棒。使用三個節點的 @ m4.2xlarge 我們將擁有 8*3 = 24 核處理器 (雖然它們是在 Spark 和 MySQL 間共享的)。我們預期可達 10 倍的性能提升，特別是在沒有覆蓋索引 (Covered index) 的情況下。

然而，在 m4.2xlarge 上的內存數量不允許我們超額運行 MySQL，因此所有的數據讀都是通過 EBS 的 IOPS，這個只能給我們提供 ~120MB/sec 的數據讀取速度。我在三臺物理機上重做了上面的測試，這三臺機器配置如下：

28 cores E5-2683 v3 @ 2.00GHz

240GB of RAM

Samsung 850 PRO

這個測試是完全在內存之外的：

Query 1 (參考上面內容)

Query 2: select dayofweek, count(*) from ontime_part group by dayofweek;

因為表只有 26 個分區，因此我們實際上並沒有足夠的併發來跑滿所有的處理器核以及內存。我也測試了帶 ID 主鍵的非分區表以及使用 128 個分區的情況。

分區的注意點

在測試中我們使用了分區表（根據年份進行分區）來降低 MySQL 資源爭用的情況。同時 Spark 中的 “partitionColumn” 參數並不要求 MySQL 表是分區的。舉例來說，如果一個表包含一個主鍵，我們可以在 Spark 中使用 CREATE VIEW：

假設我們有足夠的 MySQL 服務器（例如很多節點或者從節點），我們可以提升分區數來提升併發性（而不是根據年份分區只得到 26 個分區）。實際上，上述的測試可以得到更快的響應時間，查詢1只執行了 6.44 秒。

Spark 表現不好的地方

對於那些本身執行很快的查詢（那些完全使用索引或者是可以有效使用索引）來說，使用 Spark 沒什麼意義。因為從 MySQL 上加載數據並載入到 Spark 的開銷很大。這個開銷會影響查詢的速度。例如類似這樣的查詢

一個更好的例子是 select max(id) from ontime_part. 在 MySQL 中這個查詢會使用索引，所有的計算都在 MySQL 中完成。而如果是 Spark 則需要從 MySQL 獲取所有的 ID (select id from ontime_part) 然後計算最大值，這個查詢在 Spark 上花了 24.267 秒！

結論

在 MySQL 上使用 Apache Spark 作為一個額外的引擎層，可以幫助提升一些很慢的報表查詢的執行速度，為運行那些需要很長時間的查詢提供急需的伸縮性。此外，Spark 還可以幫我們實現常用查詢的緩存。

關注愛可生雲數據庫，交流更多數據庫技術。

閱讀更多 愛可生雲數據庫 的文章

關鍵字: Spark 類似 中央處理器