AWS Redshift ETL的幾個性能最佳實踐

Amazon Redshift 是一個支持SQL查詢的、快速、可擴展的列式存儲數據庫，它支持PB級的數量查詢，是適用于企業級的數據倉庫。同時Redshift支持大規模并發查詢、支持結果集緩存，響應查詢時間最快至亞秒，比起其他數據倉庫快將近十倍。借助 Redshift，您的等待時間更少，可將更多時間用于獲取數據見解。

ETL在計算機領域是一個很流行的概念，意指將數據從一個或多個源頭復制到目標系統的一個過程，其中包含三個步驟：
1，Extract 從數據源中選擇/提取需要導出的數據
2，Transform 將導出的數據根據業務需要進行必要的格式/表現形式上的轉換
3，Load 將轉換后的數據導入目標系統

在使用Redshift之前需要將數據導入Redshift，即Redshift的ETL。例如數據庫的遷移，將舊數據庫中的數據轉移到Redshift等等。
本文旨在分享我們bosicloud在日常工作中關于Redshift ETL方面的一些技巧及建議：

1，使用COPY命令將多個、大小相當的文件加載到Redshift
Amazon Redshift是一個MPP數據庫，即大規模并行處理數據庫，Redshift的背后是一個EC2集群，每個計算節點（ec2）進一步細分為slice，所有slice平分計算節點的處理能力。 每個節點的slice數量取決于群集的節點類型。 例如，每個DS2.XLARGE計算節點都有兩個slice，而每個DS2.8XLARGE計算節點有16個slice。

進行加載數據工作時，最好的情況是整個工作量平分給所有的計算節點（EC2 node）的slice。當只加載一個大文件，或者加載多個大小差異較大的文件時，都可能導致計算節點工作量分配不均等，進而導致整個Redshift加載數據時性能低下。例如，我們Redshift有兩個計算節點，在加載一個大文件時每個節點的工作負載如下：

從上圖可以看到，加載的任務落在了compute-0頭上，而compute-1則是閑置的，這是一個木桶原理，Redshift最終花費的時間等于工作時間最長的那個計算節點。
所以我們最好將大文件切割為多個大小相同的小文件，并且文件總數量正好是計算節點的整倍數，從而每個計算節點可以分到數目相同的小文件。另外，我們還建議將這些小文件進行壓縮，例如gzip, lzop, or bzip2因為Redshift加載及存儲數據時都支持這幾種壓縮格式。顯然，壓縮后的數據更小了，加載的工作量也就更小了。

2，使用workload management合理調整Redshift queue
Redshift通過workload management（WLM）管理著多個queue，用戶提交SQL查詢任務到Redshift時，SQL會根據提交者所在的group被分派到group對應的queue排隊等候執行。Redshift的內存及計算能力被分成許多個單元/單位，一個slot代表一個單元的內存及計算能力，一個slot同一時間可以執行一個SQL查詢任務，不同的queue擁有不同數量的slot，slot的數量決定該queue能夠同時并發執行多少個SQL查詢任務。因為ETL往往伴隨著許多COMMIT 操作，而COMMIT 都很消耗計算資源。為了進行ETL的同時不影響普通用戶提交SQL query，我們bosicloud建議將ETL和普通用戶提交的SQL query分發到不同的queue中。否則普通用戶query可能由于等待ETL COMMIT導致不能及時響應。

另外，為了加快Redshift的COPY ETL過程，我們還可以通過wlm_query_slot_count參數調整ETL queue的slot數量，從而增加queue的內存、計算能力及SQL查詢并發數量。

                                             Redshift的WLM設置界面

3，使用”BEGIN…COMMIT”減少COMMIT次數前面我們提到ELT是一個多步驟的任務，每個步驟最后往往需要執行一個COMMIT，而COMMIT又是一個昂貴的操作。所以我們bosicloud建議使用BEGIN…END 將能夠合并的多個步驟盡量合并為一個步驟并只執行一次COMMIT，例如：
Begin
CREATE temporary staging_table;
INSERT INTO staging_table SELECT .. FROM source;
DELETE FROM table1 WHERE ???=???;
INSERT INTO table1 SELECT .. FROM staging_table;
DELETE FROM table2 WHERE ???=???;
INSERT INTO table2 SELECT .. FROM staging_table;
Commit

4，使用Redshift Spectrum for ad-hoc查詢
在以往，我們可以會為了做ad-hoc查詢而將S3上的數據ETL加載到Redshift進行查詢。如果僅為了偶爾的一兩次查詢而進行ETL，這個ETL將顯得非常昂貴、不劃算。別忘了AWS最近推出了Redshift Spectrum新功能，即您可以直接利用Redshift Spectrum查詢S3上的數據而無需將數據加載到Redshift中，雖然Specturm查詢相比普通Redshift查詢較慢，但比起ETL來說查詢速度仍然是大大提升了。

5，關于ETL健康檢查的SQL實用腳本：
? 返回過去1天內queue的統計信息，例如最大隊列長度和隊列時間
select startqueue,node, datediff(ms,startqueue,startwork) as queue_time, datediff(ms, startwork, endtime) as commit_time, queuelen
from stl_commit_stats
where startqueue >= dateadd(day, -1, current_Date)
order by queuelen desc , queue_time desc;
? 返回一個星期內所執行的COPY的相關信息，如COPY的開始時間（Starttime），所在queue的ID(query),SQL語句（querytxt），COPY的文件數量（n_files）及文件大小（size_mb）等等：
select q.starttime, s.query, substring(q.querytxt,1,120) as querytxt,
s.n_files, size_mb, s.time_seconds,
s.size_mb/decode(s.time_seconds,0,1,s.time_seconds) as mb_per_s
from (select query, count() as n_files,
sum(transfer_size/(10241024)) as size_MB, (max(end_Time) -
min(start_Time))/(1000000) as time_seconds , max(end_time) as end_time
from stl_s3client where http_method = 'GET' and query > 0
and transfer_time > 0 group by query ) as s
LEFT JOIN stl_Query as q on q.query = s.query
where s.end_Time >= dateadd(day, -7, current_Date)
order by s.time_Seconds desc, size_mb desc, s.end_time desc
limit 50;

? 建立view視圖查看每個表空間使用情況，請考慮將空間增長較快的表的內容unload到S3.

CREATE OR REPLACE VIEW admin.v_space_used_per_tbl
AS with info_table as ( SELECT TRIM(pgdb.datname) AS dbase_name
,TRIM(pgn.nspname) as schemaname
,TRIM(pgc.relname) AS tablename
,id AS tbl_oid
,b.mbytes AS megabytes
,CASE WHEN pgc.reldiststyle = 8
THEN a.rows_all_dist
ELSE a.rows END AS rowcount
,CASE WHEN pgc.reldiststyle = 8
THEN a.unsorted_rows_all_dist
ELSE a.unsorted_rows END AS unsorted_rowcount
,CASE WHEN pgc.reldiststyle = 8
THEN decode( det.n_sortkeys,0, NULL,DECODE( a.rows_all_dist,0,0, (a.unsorted_rows_all_dist::DECIMAL(32)/a.rows_all_dist)100))::DECIMAL(20,2)
ELSE decode( det.n_sortkeys,0, NULL,DECODE( a.rows,0,0, (a.unsorted_rows::DECIMAL(32)/a.rows)100))::DECIMAL(20,2) END
AS pct_unsorted
FROM ( SELECT
db_id
,id
,name
,MAX(ROWS) AS rows_all_dist
,MAX(ROWS) - MAX(sorted_rows) AS unsorted_rows_all_dist
,SUM(rows) AS rows
,SUM(rows)-SUM(sorted_rows) AS unsorted_rows
FROM stv_tbl_perm
GROUP BY db_id, id, name
) AS a
INNER JOIN
pg_class AS pgc
ON pgc.oid = a.id
INNER JOIN
pg_namespace AS pgn
ON pgn.oid = pgc.relnamespace
INNER JOIN
pg_database AS pgdb
ON pgdb.oid = a.db_id
INNER JOIN (SELECT attrelid,
MIN(CASE attisdistkey WHEN 't' THEN attname ELSE NULL END) AS "distkey",
MIN(CASE attsortkeyord WHEN 1 THEN attname ELSE NULL END) AS head_sort,
MAX(attsortkeyord) AS n_sortkeys,
MAX(attencodingtype) AS max_enc,
SUM(case when attencodingtype <> 0 then 1 else 0 end)::DECIMAL(20,3)/COUNT(attencodingtype)::DECIMAL(20,3) 100.00 as pct_enc
FROM pg_attribute
GROUP BY 1) AS det ON det.attrelid = a.id
LEFT OUTER JOIN
( SELECT
tbl
,COUNT() AS mbytes
FROM stv_blocklist
GROUP BY tbl
) AS b
ON a.id=b.tbl
WHERE pgc.relowner > 1)
select info.*
,CASE WHEN info.rowcount = 0 THEN 'n/a'
WHEN info.pct_unsorted >= 20 THEN 'VACUUM SORT recommended'
ELSE 'n/a'
END AS recommendation
from info_table info;

? 找出本周內最費時間的前50個SQL查詢（多個相同的sql查詢時間合并計算）
-- query runtimes
select trim(database) as DB, count(query) as n_qry, max(substring (qrytext,1,80)) as qrytext, min(run_seconds) as "min" , max(run_seconds) as "max", avg(run_seconds) as "avg", sum(run_seconds) as total, max(query) as max_query_id,
max(starttime)::date as last_run, aborted,
listagg(event, ', ') within group (order by query) as events
from (
select userid, label, stl_query.query, trim(database) as database, trim(querytxt) as qrytext, md5(trim(querytxt)) as qry_md5, starttime, endtime, datediff(seconds, starttime,endtime)::numeric(12,2) as run_seconds,
aborted, decode(alrt.event,'Very selective query filter','Filter','Scanned a large number of deleted rows','Deleted','Nested Loop Join in the query plan','Nested Loop','Distributed a large number of rows across the network','Distributed','Broadcasted a large number of rows across the network','Broadcast','Missing query planner statistics','Stats',alrt.event) as event
from stl_query
left outer join ( select query, trim(split_part(event,':',1)) as event from STL_ALERT_EVENT_LOG where event_time >= dateadd(day, -7, current_Date) group by query, trim(split_part(event,':',1)) ) as alrt on alrt.query = stl_query.query
where userid <> 1
-- and (querytxt like 'SELECT%' or querytxt like 'select%' )
-- and database = ''
and starttime >= dateadd(day, -7, current_Date)
)
group by database, label, qry_md5, aborted
order by total desc limit 50;

【關于博思云為】
作為一家專業的云計算服務型企業，博思云為專為客戶提供 AWS 上的運營服務：包括架構咨詢服務、遷移服務、云安全集成服務、混合云管理服務、大數據服務以及 DevOps 服務。目前，博思云為在大數據、DevOps、架構、數據庫以及操作系統等都已取得廠商認證，在上海、南京、杭州、武漢等地設有分公司。為創新服務模式、引領 IT 服務業的發展，博思云為將持續投入資源開展智能混合云管理平臺、圖數據庫的研發等。