選擇 Parquet for Spark SQL 的 5 大原因

2016-03-01 11:11:38

Parquet 用于 Spark SQL 時表現非常出色。它不僅提供了更高的壓縮率，還允許通過已選定的列和低級別的讀取器過濾器來只讀取感興趣的記錄。本文將為您詳細介紹使用 Parquet for Spark SQL 優勢的 5 大原因。

列式存儲 (columnar storage) 在處理大數據的時候可以有效地節省時間和空間。例如，與使用文本相比，Parquet 讓 Spark SQL 的性能平均提高了 10 倍，這要感謝初級的讀取器過濾器、高效的執行計劃，以及 Spark 1.6.0 中經過改進的掃描吞吐量!本文將為您詳細介紹使用 Parquet for Spark SQL 優勢的 5 大原因。

為了了解 Parquet 有多么強大，我們從 spark-perf-sql 中挑選了 24 個從 TPC-DS 中衍生的查詢來完成比較(總共有 99 個查詢，一些查詢在 1TB 的縮放比例下無法用于平面的 CSV 數據文件。更多內容參見下文)。這些查詢代表了 TPC-DS 中的所有類別：報告、即席報告、迭代和數據挖掘。我們還要確保包含了短查詢(查詢12 和 91)和長時間運行的查詢(查詢 24a 和 25)，以及會使用 100% CPU 的眾所周知的查詢(查詢 97)。

我們使用了一個 6 節點的預置型 Cisco UCS 集群，每個 Cisco 驗證了的設計都有類似的配置。我們調優了底層硬件，以防在所有測試中遇到網絡或磁盤 IO 瓶頸。本文的重點是了解在 Spark 1.5.1 和剛發布的 Spark 1.6.0 中只對文本和 Parquet 存儲格式運行這些查詢會有怎樣的性能差異。總的 Spark 工作存儲為 500GB。TPC-DS 縮放比例為 1TB。

1. Spark SQL 在用于 Parquet 時更快一些!

下圖比較了在 Spark 1.5.1 中運行 24 個查詢的所有執行時間的總和。在使用平面的 CVS 文件時，查詢花費了大約 12 個小時才完成，而在使用 Parquet 時，查詢用了不到 1 個小時的時間就完成了，性能提高了 11 倍。

圖 1. 比較在文本和 Parquet 中花費的總查詢時間(以秒為單位)，越小越好。

2. Spark SQL 在使用較大縮放比例時的表現要優于 Parquet

存儲格式的選擇不當往往會導致難以診斷和難以修復。例如，在采用 1TB 的縮放比例時，如果使用平面 CSV 文件，在所有可運行的查詢中，至少有 1/3 的查詢無法完成，但在使用 Parquet 時，這些查詢都完成了。

一些錯誤和異常非常神秘。這里有 3 個示例：

錯誤示例 1：

WARN scheduler.TaskSetManager: Lost task 145.0 in stage 4.0 (TID 4988, rhel8.cisco.com): FetchFailed(BlockManagerId(2, rhel4.cisco.com, 49209), shuffleId=13, mapId=47, reduceId=145, message= 
         org.apache.spark.shuffle.FetchFailedException: java.io.FileNotFoundException: /data6/hadoop/yarn/local/usercache/spark/appcache/application_1447965002296_0142/blockmgr-44627d4c-4a2b-4f53-a471-32085a252cb0/15/shuffle_13_119_0.index (No such file or directory) 
         at java.io.FileInputStream.open0(Native Method) 
         at java.io.FileInputStream.open(FileInputStream.java:195)

錯誤示例 2：

WARN scheduler.TaskSetManager: Lost task 1.0 in stage 13.1 (TID 13621, rhel7.cisco.com): FetchFailed(null, shuffleId=9, mapId=-1, reduceId=148, message= 
            org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 9 
            at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:460) 
            at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:456) 
            at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772) 
            at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) 
            at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)

錯誤示例 3：

ERROR cluster.YarnScheduler: Lost executor 59 on rhel4.cisco.com: remote Rpc client disassociated

大多數查詢的失敗迫使 Spark 通過重新排隊任務(甚至是重新啟動某個階段)來進行再次嘗試。事情從那時起變得更糟;最終，該應用程序失敗了，像是永遠不會完成。

通過切換到 Parquet，無需更改其他任何 Spark 配置，這些問題就得到了解決。壓縮減小了文件的大小，列式格式允許只讀取選擇的記錄，減少的輸入數據直接影響了 Spark DAG 調度程序關于執行圖的決策(更多的細節參見下文)。Parquet 的所有這些優勢都對查詢的快速完成至關重要。

3. 更少的磁盤 IO

采用了壓縮功能的 Parquet 能夠讓數據存儲平均減少 75%，也就是說，1TB 壓縮比例的數據文件在磁盤上只會占用大約 250 GB 的磁盤空間。這顯著減少了 Spark SQL 應用程序所需的輸入數據。而且在 Spark 1.6.0 中，Parquet 讀取器使用了下推過濾器來進一步減少磁盤 IO。下推式過濾器允許在將數據讀入 Spark 之前就制定數據選擇決策。例如，對查詢 97 中的 between 子句的處理如下所示：

select cs_bill_customer_sk customer_sk, cs_item_sk item_sk
from catalog_sales,date_dim
where cs_sold_date_sk = d_date_sk 
and d_month_seq between 1200 and 1200 + 11

Spark SQL 展示了查詢的物理計劃中的以下 scan 語句：

+- Scan ParquetRelation[d_date_sk#141,d_month_seq#144L] InputPaths: hdfs://rhel10.cisco.com/user/spark/hadoopds1tbparquet/date_dim/_SUCCESS, hdfs://rhel10.cisco.com/user/spark/hadoopds1tbparquet/date_dim/_common_metadata, hdfs://rhel10.cisco.com/user/spark/hadoopds1tbparquet/date_dim/_metadata, hdfs://rhel10.cisco.com/user/spark/hadoopds1tbparquet/date_dim/part-r-00000-4d205b7e-b21d-4e8b-81ac-d2a1f3dd3246.gz.parquet, hdfs://rhel10.cisco.com/user/spark/hadoopds1tbparquet/date_dim/part-r-00001-4d205b7e-b21d-4e8b-81ac-d2a1f3dd3246.gz.parquet, PushedFilters: [GreaterThanOrEqual(d_month_seq,1200), LessThanOrEqual(d_month_seq,1211)]]

其中，PushedFilters 只返回 d_mont_seq 列中范圍 1200 到 1211 的記錄，或者只返回幾個記錄。與平面文件相比較，在使用平面文件時，會讀取整個表(每一列和每一行)，如物理計劃中所示：

[                  Scan CsvRelation(hdfs://rhel10.cisco.com/user/spark/hadoopds1000g/date_dim/*,false,|,",null,PERMISSIVE,COMMONS,false,false,StructType(StructField(d_date_sk,IntegerType,false), StructField(d_date_id,StringType,false), StructField(d_date,StringType,true), StructField(d_month_seq,LongType,true), StructField(d_week_seq,LongType,true), StructField(d_quarter_seq,LongType,true), StructField(d_year,LongType,true), StructField(d_dow,LongType,true), StructField(d_moy,LongType,true), StructField(d_dom,LongType,true), StructField(d_qoy,LongType,true), StructField(d_fy_year,LongType,true), StructField(d_fy_quarter_seq,LongType,true), StructField(d_fy_week_seq,LongType,true), StructField(d_day_name,StringType,true), StructField(d_quarter_name,StringType,true), StructField(d_holiday,StringType,true), StructField(d_weekend,StringType,true), StructField(d_following_holiday,StringType,true), StructField(d_first_dom,LongType,true), StructField(d_last_dom,LongType,true), StructField(d_same_day_ly,LongType,true), StructField(d_same_day_lq,LongType,true), StructField(d_current_day,StringType,true), StructField(d_current_week,StringType,true), StructField(d_current_month,StringType,true), StructField(d_current_quarter,StringType,true), StructField(d_current_year,StringType,true)))[d_date_sk#141,d_date_id#142,d_date#143,d_month_seq#144L,d_week_seq#145L,d_quarter_seq#146L,d_year#147L,d_dow#148L,d_moy#149L,d_dom#150L,d_qoy#151L,d_fy_year#152L,d_fy_quarter_seq#153L,d_fy_week_seq#154L,d_day_name#155,d_quarter_name#156,d_holiday#157,d_weekend#158,d_following_holiday#159,d_first_dom#160L,d_last_dom#161L,d_same_day_ly#162L,d_same_day_lq#163L,d_current_day#164,d_current_week#165,d_current_month#166,d_current_quarter#167,d_current_year#168]]

4. Spark 1.6.0 提供了更高的掃描吞吐量

Databricks 的 Spark 1.6.0 發布博客中曾經提到過顯著的平面掃描吞吐量，因為該博客使用到了 “更優化的代碼路徑” 一詞。為了在現實世界中說明這一點，我們在 Spark 1.5.1 和 1.6.0 中運行了查詢 97，并捕獲了 nmon 數據。改進非常明顯。

首先，查詢響應時間減少了一半：查詢 97 在 Spark 1.5.1 中用了 138 秒時間，而在 Spark 1.6.0 中只用了 60 秒。