Spark鏈接到MySQL並執行查詢爲何速度會快？

問題導讀：
1. Spark爲何能提升Mysql的查詢速度？
2. 如何運行SQL in Spark？
3. SparkSQL如何將查詢推送到MySQL？
4. 如何使用Spark緩存查詢數據？ 5. 如何使用 Spark 和 Percona XtraDB Cluster？
6. Spark表分區時須要注意的事項？
7. Spark表現很差的時候？

在這篇文章中咱們將討論如何利用 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).
  

示例:

[Shell]  純文本查看  複製代碼

?

1 2 3 4 5 6

root@thor:~ /spark # ./sbin/start-master.sh less .. /logs/spark-root-org .apache.spark.deploy.master.Master-1-thor.out 15 /08/25 11:21:21 INFO Master: Starting Spark master at spark: //thor :7077 15 /08/25 11:21:21 INFO Utils: Successfully started service 'MasterUI' on port 8080. 15 /08/25 11:21:21 INFO MasterWebUI: Started MasterWebUI at [url=http: //10 .60.23.188:8080]http: //10 .60.23.188:8080[ /url ] root@thor:~ /spark # ./sbin/start-slave.sh spark://thor:7077

爲了鏈接到 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 中添加類路徑，以下：

[Shell]  純文本查看  複製代碼

?

1 2	spark.driver.extraClassPath = /usr/local/spark/mysql-connector-java-5 .1.39-bin.jar spark.executor.extraClassPath = /usr/local/spark/mysql-connector-java-5 .1.39-bin.jar

利用 Apache Spark 運行 MySQL 查詢

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

如今咱們能夠在 Spark 中運行 MySQL 查詢了。首先，從 Spark 目錄中啓動 Shell （在我這裏是 /usr/local/spark ）：

[Shell]  純文本查看  複製代碼

?

1	$ . /bin/spark-shell --driver-memory 4G --master spark: //server1 :7077

而後咱們將鏈接到 MySQL 服務器並註冊臨時視圖：

[Scala]  純文本查看  複製代碼

?

1 2 3 4 5 6 7

val jdbcDF = spark.read.format( "jdbc" ).options(    Map( "url" ->  "jdbc:mysql://localhost:3306/ontime?user=root&password=" ,    "dbtable" -> "ontime.ontime_part" ,    "fetchSize" -> "10000" ,    "partitionColumn" -> "yeard" , "lowerBound" -> "1988" , "upperBound" -> "2016" , "numPartitions" -> "28"    )).load() jdbcDF.createOrReplaceTempView( "ontime" )

這樣咱們就爲 Spark 建立了一個「數據源」（換句話說就是至關於 Spark 創建了到 MySQL 的鏈接）。Spark 表名 「ontime」 對應鏈接到 MySQL 的ontime.ontime_part 表，如今能夠在 Spark 中運行 SQL 了，它們是按順序被一一解析並轉換成 MySQL 查詢的。

「 partitionColumn」 在這裏很是重要，它告訴 Spark 並行的執行多個查詢，每一個分區分配一個查詢執行。

如今咱們能夠運行查詢：

[Scala]  純文本查看  複製代碼

?

1 2

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()

MySQL 查詢示例

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

[SQL]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11

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' ) GROUP by carrier HAVING cnt > 100000 and max_year > '1990' ORDER by rate DESC , cnt desc LIMIT  10

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

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

[SQL]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18

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 | + -------------------------------------+

表結構以下：

[SQL]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

CREATE TABLE `ontime_part` (    `YearD` int (11) NOT NULL ,    `Quarter` tinyint(4) DEFAULT NULL ,    `MonthD` tinyint(4) DEFAULT NULL ,    `DayofMonth` tinyint(4) DEFAULT NULL ,    `DayOfWeek` tinyint(4) DEFAULT NULL ,    `FlightDate` date DEFAULT NULL ,    `UniqueCarrier` char (7) DEFAULT NULL ,    `AirlineID` int (11) DEFAULT NULL ,    `Carrier` char (2) DEFAULT NULL ,    `TailNum` varchar (50) DEFAULT NULL , ...    `id` int (11) NOT NULL AUTO_INCREMENT,    PRIMARY KEY (`id`,`YearD`),    KEY `covered` (`DayOfWeek`,`OriginState`,`DestState`,`Carrier`,`YearD`,`ArrDelayMinutes`) ) ENGINE=InnoDB AUTO_INCREMENT=162668935 DEFAULT CHARSET=latin1 /*!50100 PARTITION BY RANGE (YearD) (PARTITION p1987 VALUES LESS THAN (1988) ENGINE = InnoDB,   PARTITION p1988 VALUES LESS THAN (1989) ENGINE = InnoDB,   PARTITION p1989 VALUES LESS THAN (1990) ENGINE = InnoDB,   PARTITION p1990 VALUES LESS THAN (1991) ENGINE = InnoDB,   PARTITION p1991 VALUES LESS THAN (1992) ENGINE = InnoDB,   PARTITION p1992 VALUES LESS THAN (1993) ENGINE = InnoDB,   PARTITION p1993 VALUES LESS THAN (1994) ENGINE = InnoDB,   PARTITION p1994 VALUES LESS THAN (1995) ENGINE = InnoDB,   PARTITION p1995 VALUES LESS THAN (1996) ENGINE = InnoDB,   PARTITION p1996 VALUES LESS THAN (1997) ENGINE = InnoDB,   PARTITION p1997 VALUES LESS THAN (1998) ENGINE = InnoDB,   PARTITION p1998 VALUES LESS THAN (1999) ENGINE = InnoDB,   PARTITION p1999 VALUES LESS THAN (2000) ENGINE = InnoDB,   PARTITION p2000 VALUES LESS THAN (2001) ENGINE = InnoDB,   PARTITION p2001 VALUES LESS THAN (2002) ENGINE = InnoDB,   PARTITION p2002 VALUES LESS THAN (2003) ENGINE = InnoDB,   PARTITION p2003 VALUES LESS THAN (2004) ENGINE = InnoDB,   PARTITION p2004 VALUES LESS THAN (2005) ENGINE = InnoDB,   PARTITION p2005 VALUES LESS THAN (2006) ENGINE = InnoDB,   PARTITION p2006 VALUES LESS THAN (2007) ENGINE = InnoDB,   PARTITION p2007 VALUES LESS THAN (2008) ENGINE = InnoDB,   PARTITION p2008 VALUES LESS THAN (2009) ENGINE = InnoDB,   PARTITION p2009 VALUES LESS THAN (2010) ENGINE = InnoDB,   PARTITION p2010 VALUES LESS THAN (2011) ENGINE = InnoDB,   PARTITION p2011 VALUES LESS THAN (2012) ENGINE = InnoDB,   PARTITION p2012 VALUES LESS THAN (2013) ENGINE = InnoDB,   PARTITION p2013 VALUES LESS THAN (2014) ENGINE = InnoDB,   PARTITION p2014 VALUES LESS THAN (2015) ENGINE = InnoDB,   PARTITION p2015 VALUES LESS THAN (2016) ENGINE = InnoDB,   PARTITION p_new VALUES LESS THAN MAXVALUE ENGINE = InnoDB) */

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

[SQL]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11 12 13

mysql>  explain 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  10G *************************** 1. row ***************************             id: 1    select_type: SIMPLE          table : ontime_part           type: range possible_keys: covered            key : covered        key_len: 2            ref: NULL           rows : 70483364          Extra: Using where ; Using index ; Using temporary ; Using filesort 1 row in set (0.00 sec)

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

[SQL]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16

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; + ------------+----------+---------+----------+-----------------+------+ | min (yearD) | max_year | Carrier | cnt      | flights_delayed | rate | + ------------+----------+---------+----------+-----------------+------+ |       2003 |     2013 | EV      |  2962008 |          464264 | 0.16 | |       2003 |     2013 | B6      |  1237400 |          187863 | 0.15 | |       2006 |     2011 | XE      |  1615266 |          230977 | 0.14 | |       2003 |     2005 | DH      |   501056 |           69833 | 0.14 | |       2001 |     2013 | MQ      |  4518106 |          605698 | 0.13 | |       2003 |     2013 | FL      |  1692887 |          212069 | 0.13 | |       2004 |     2010 | OH      |  1307404 |          175258 | 0.13 | |       2006 |     2013 | YV      |  1121025 |          143597 | 0.13 | |       2003 |     2006 | RU      |  1007248 |          126733 | 0.13 | |       1988 |     2013 | UA      | 10717383 |         1327196 | 0.12 | + ------------+----------+---------+----------+-----------------+------+ 10 rows in set (19 min 16.58 sec)

足足執行了 19 分鐘，這個結果然的讓人爽不起來。

SQL in Spark

如今咱們但願在 Spark 中運行相同的查詢，讓 Spark 從 MySQL 讀取數據。咱們建立了一個「數據源」而後執行以下查詢：

[Scala]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

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() +----------+--------+-------+--------+---------------+----+ |min(yearD)|max _ year|Carrier|     cnt|flights _ delayed|rate| +----------+--------+-------+--------+---------------+----+ |      2003 |    2013 |     EV| 2962008 |         464264 | 0.16 | |      2003 |    2013 |     B 6 | 1237400 |         187863 | 0.15 | |      2006 |    2011 |     XE| 1615266 |         230977 | 0.14 | |      2003 |    2005 |     DH|  501056 |          69833 | 0.14 | |      2001 |    2013 |     MQ| 4518106 |         605698 | 0.13 | |      2003 |    2013 |     FL| 1692887 |         212069 | 0.13 | |      2004 |    2010 |     OH| 1307404 |         175258 | 0.13 | |      2006 |    2013 |     YV| 1121025 |         143597 | 0.13 | |      2003 |    2006 |     RU| 1007248 |         126733 | 0.13 | |      1988 |    2013 |     UA| 10717383 |        1327196 | 0.12 | +----------+--------+-------+--------+---------------+----+

Spark-shell 並不會顯示查詢的執行時間，這個能夠從 spark-sql 提供的 Web UI 中獲取到。我在 spark-sql 中從新執行相同的查詢：

[Shell]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

. /bin/spark-sql --driver-memory 4G  --master spark: //thor :7077 spark-sql> CREATE TEMPORARY VIEW ontime           > USING org.apache.spark.sql.jdbc           > OPTIONS (           >      url  "jdbc:mysql://localhost:3306/ontime?user=root&password=" ,           >      dbtable "ontime.ontime_part" ,           >      fetchSize "1000" ,           >      partitionColumn "yearD" , lowerBound "1988" , upperBound "2014" , numPartitions "48"           > ); 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 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. 2003    2013    EV      2962008 464264  0.16 2003    2013    B6      1237400 187863  0.15 2006    2011    XE      1615266 230977  0.14 2003    2005    DH      501056  69833   0.14 2001    2013    MQ      4518106 605698  0.13 2003    2013    FL      1692887 212069  0.13 2004    2010    OH      1307404 175258  0.13 2006    2013    YV      1121025 143597  0.13 2003    2006    RU      1007248 126733  0.13 1988    2013    UA      10717383        1327196 0.12 Time taken: 139.628 seconds, Fetched 10 row(s)

能夠看到查詢的時間足足快了 10 倍之多（同一臺機器，只有一臺機器）。可是到底這些查詢是怎麼變成 MySQL 查詢的呢？而後爲何這樣的查詢會快那麼多。讓咱們深刻到 MySQL 一探究竟。

深刻 MySQL

Spark:

[Shell]  純文本查看  複製代碼

?

1 2	scala> sqlDF.show() [Stage 4:>                                                        (0 + 26) / 26]

MySQL:

[Shell]  純文本查看  複製代碼

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32

mysql> select id , info from information_schema.processlist where info is not NULL and info not like '%information_schema%' ; +-------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | id    | info                                                                                                                                                                                                                                                    | +-------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | 10948 | 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) | | 10965 | 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 >= 2007 AND yearD < 2008) | | 10966 | 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 >= 1991 AND yearD < 1992) | | 10967 | 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 >= 1994 AND yearD < 1995) | | 10968 | 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 >= 1998 AND yearD < 1999) | | 10969 | 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 >= 2010 AND yearD < 2011) | | 10970 | 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 >= 2002 AND yearD < 2003) | | 10971 | 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 >= 2006 AND yearD < 2007) | | 10972 | 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 >= 1990 AND yearD < 1991) | | 10953 | 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 >= 2009 AND yearD < 2010) | | 10947 | SELECT `YearD`,`ArrDelayMinutes`,`Carrier` FROM ontime.ontime_part WHERE (((NOT (DayOfWeek IN (6, 7)))) AND ((NOT (OriginState IN ( 'AK' , 'HI' ,