原Greenplum集群在進(jìn)行10億*1億的關(guān)聯(lián)查詢時達(dá)到極限，無法支撐更大數(shù)據(jù)量級的關(guān)聯(lián)查詢。使用DorisDB替換Greenplum構(gòu)建新的集群，在進(jìn)行736億*15億的超大量級數(shù)據(jù)關(guān)聯(lián)查詢時，不僅可以順利完成，并且耗時很短，對業(yè)務(wù)的整體提升巨大。

　　一、使用背景

　　1.1選用原因

　　我司原有業(yè)務(wù)查詢使用的數(shù)據(jù)庫為Greenplum，在數(shù)據(jù)源變更后，數(shù)據(jù)量從原來的日增千萬級別(近百G)暴增至日增千億(10T)級別，原有的12臺GP集群在數(shù)據(jù)量增長后存在以下痛點：

　　1、數(shù)據(jù)導(dǎo)入

　　原有的數(shù)據(jù)導(dǎo)入借助于gpload的工具，在有索引的情況下，數(shù)據(jù)導(dǎo)入隨著數(shù)據(jù)量的增加會變慢，在千億級日增情況下，有索引的表根本無法導(dǎo)入。即使使用先導(dǎo)入數(shù)據(jù)，后建索引的方式，導(dǎo)入過程還是不理想，建索引的時間會由于數(shù)據(jù)量的增長而增長，由于機器資源在現(xiàn)有的基礎(chǔ)上增加的的可能性不是很大，使用該方式做數(shù)據(jù)導(dǎo)入，整個流程耗時相當(dāng)長，無法滿足業(yè)務(wù)需求。

　　2、數(shù)據(jù)存儲

　　GP在數(shù)據(jù)存儲這一塊，如果使用heap表的方式創(chuàng)建表，數(shù)據(jù)來說是不做任何壓縮進(jìn)行存儲，比較占用存儲資源。如果采用列存表的方式，需要手動指定壓縮等級和字段，但是使用者在不清楚數(shù)據(jù)重復(fù)的具體情況下設(shè)置該參數(shù)就只能是想當(dāng)然去做，然后在生產(chǎn)中根據(jù)數(shù)據(jù)實際情況進(jìn)行更改，在查詢時，cpu會進(jìn)行解壓縮操作，增加了cpu的計算耗時。

　　3、數(shù)據(jù)計算

　　計算瓶頸其實是我們數(shù)據(jù)量增長之后主要的痛點。在原有的使用過程中，針對于業(yè)務(wù)A的整體運行時長，從客戶觸發(fā)到最終顯示，需要大概100分鐘左右，數(shù)據(jù)量增長后，業(yè)務(wù)A的基本跑不動。其次，在日常的etl過程中，一些定時表關(guān)聯(lián)在原GP的處理過程中只能是對事實表按照時間粒度做切分，小部分小部分的進(jìn)行關(guān)聯(lián)，然后再進(jìn)行合并處理，數(shù)據(jù)量增長之后的關(guān)聯(lián)，在現(xiàn)有資源下也無法實現(xiàn)。

　　在GP無法承受如此巨大的數(shù)據(jù)量，滿足不了業(yè)務(wù)的需求時，我們將目光轉(zhuǎn)向其他解決方案，在測試了DorisDB，clickhouse以及其他olap產(chǎn)品后，結(jié)合自身的業(yè)務(wù)特點和使用上的易用性，最終選用了DorisDB作為MPP的解決方案。此文檔也是基于DorisDB進(jìn)行詳細(xì)的業(yè)務(wù)測試過程中整理的文檔。

　　1.2集群配置

　　此次測試使用的機器資源如下所示(只部署了DorisDB的環(huán)境)：

　　機器數(shù)量：10臺

　　機器系統(tǒng)：centos7.6

　　機器內(nèi)存：256G

　　機器磁盤：7200轉(zhuǎn)機械硬盤，每臺機器為8T*4,做了raid0

　　網(wǎng)絡(luò)帶寬：內(nèi)網(wǎng)萬兆光遷

　　CPU:2*12 core

　　此次部署的DorisDB的集群詳情如下(未使用spark load，沒有安裝spark的客戶端)：

　　fe數(shù)量：3臺(1 master+2 follower)

　　be數(shù)量：10臺

　　broker數(shù)量：10臺

　　1.3集群配置參數(shù)

　　針對自身業(yè)務(wù)特點，修改了以下參數(shù)：

　　fe:

　　broker load的參數(shù)

　　1.允許運行的最大的broker數(shù)量

　　max_broker_concurrency=10

　　2.每個be處理的數(shù)據(jù)量

　　max_bytes_per_broker_scanner=32212254720

　　上述兩個參數(shù)影響broker load導(dǎo)入時be處理數(shù)據(jù)的并發(fā)數(shù)量和單個be處理的數(shù)據(jù)量

　　be:

　　文件合并的參數(shù)

　　1.be節(jié)點base compaction線程數(shù)量

　　base_compaction_num_threads_per_disk=4

　　2.base compaction時寫磁盤的限速，單位為M

　　base_compaction_write_mbytes_per_sec=20

　　3.be節(jié)點cumulative compaction的線程數(shù)量

　　cumulative_compaction_num_threads_per_disk=8

　　4.be節(jié)點cumulative compactiond寫磁盤的限速，單位為M

　　cumulative_compaction_write_mbytes_per_sec=300

　　5.be節(jié)點cumulative compactiond線程輪詢的間隔

　　cumulative_compaction_check_interval_seconds=2

　　上述五個參數(shù)主要控制DorisDB對于文件合并的效率，可以根據(jù)自身的硬件性能和實際業(yè)務(wù)情況調(diào)整該參數(shù)。大量數(shù)據(jù)導(dǎo)入到DorisDB中時，DorisDB需要根據(jù)排序key做排序，根據(jù)字段的值做壓縮合并的操作，此時會占用磁盤性能，調(diào)整該參數(shù)(業(yè)務(wù)閑時)可以加速這一過程，使DorisDB專注于計算。

　　以上參數(shù)僅提供參考，請根據(jù)自身資源和實際情況酌情調(diào)整

　　二、數(shù)據(jù)導(dǎo)入

　　由于數(shù)據(jù)源的特殊性，數(shù)據(jù)存放在文件中，原始文件為壓縮文件，因此在實際測試過程中，我們主要對以下幾種導(dǎo)入進(jìn)行了測試(spark load未測試成功)，最終選取了broker load的方式作為最終的數(shù)據(jù)導(dǎo)入的方案。該方案能夠?qū)崿F(xiàn)單任務(wù)200W+/s的導(dǎo)入速度，并且支持并行的方式，進(jìn)一步提高數(shù)據(jù)導(dǎo)入速度。

　　2.1 stream load

　　剛開始使用DorisDB時，我們使用的導(dǎo)入方式即為stream load的方式測試小批量的數(shù)據(jù)，但是在數(shù)據(jù)量增大的情況下，大概數(shù)據(jù)單次導(dǎo)入到100G時，發(fā)現(xiàn)這種微批導(dǎo)入的方式有數(shù)據(jù)膨脹的情況，導(dǎo)入前的數(shù)據(jù)和入庫后的數(shù)據(jù)量對比差異明顯(導(dǎo)入前100G左右，導(dǎo)入后DorisDB在250G左右，并且磁盤IO占用高居不下)，遂放棄。

　　2.2 datax

　　datax主要是由于有豐富的使用經(jīng)驗，其次是datax在對于數(shù)據(jù)接入過程中很靈活，可以增加很多豐富的transformer插件來減輕后續(xù)的數(shù)據(jù)清洗的壓力。datax使用時我們主要使用的是mysqlWriter和利用stream load實現(xiàn)的DorisWriter(此處艾特社區(qū)張懷北同學(xué))。前者在我們測試時，DorisDB文檔中還未增加doriswriter的內(nèi)容，利用的是mysql的jdbc連接實現(xiàn)數(shù)據(jù)的導(dǎo)入。后者則是社區(qū)利用stream load的api實現(xiàn)的數(shù)據(jù)導(dǎo)入。在使用過程中，發(fā)現(xiàn)導(dǎo)入速度并不理想(10臺一起跑，導(dǎo)入速度在60W/s)，滿足不了我們每天的增量數(shù)據(jù)的導(dǎo)入的要求。也放棄datax的導(dǎo)入方案。(如果有對這種方式感興趣的同學(xué)可以在社區(qū)留言)。文章末尾附件有該writer實現(xiàn)的核心代碼。

　　2.3 broker load

　　broker load的導(dǎo)入方式是我們最終采用的方案。原本對spark load的方式抱有很大希望，因為我們業(yè)務(wù)中的數(shù)據(jù)另一個導(dǎo)入方向為hbase，使用的導(dǎo)入方式為bulkload的方式，利用spark合成Hfile的方式寫入hbase，該方式能夠?qū)⒋龑?dǎo)入的數(shù)據(jù)進(jìn)行排序后，形成hbase底層需要的hfile的格式寫入到hdfs，hbase可以不用再將數(shù)據(jù)在內(nèi)存中排序后再落盤，在進(jìn)行合并形成hfile，能夠借助于spark計算集群減輕hbase排序和文件合并的壓力，使得hbase專注于業(yè)務(wù)。我們猜想DorisDB的spark load是否也采用了類似的思想，利用spark處理數(shù)據(jù)后直接生成DorisDB所需要的底層存儲文件后寫入DorisDB，但是在經(jīng)過咨詢后，現(xiàn)有的spark load不具備這種提前排序生成底層存儲文件的導(dǎo)入功能，但是在未來會開發(fā)。后續(xù)開發(fā)完成后，對于DorisDB的導(dǎo)入應(yīng)該提升很大(個人臆想_)。

　　broker load的時候我們測試了分別從hdfs load csv文件和parquet文件，最終發(fā)現(xiàn)使用parquet導(dǎo)入比csv性能高出兩倍到三倍的樣子(相同數(shù)據(jù)條數(shù)，字段)，也剛好是parquet文件和csv文件實際存儲相差的樣子。同時在導(dǎo)入時，可以先將待導(dǎo)入的表的副本設(shè)為1，可以減少導(dǎo)入過程中的數(shù)據(jù)clone，加快導(dǎo)入速度。最終測的的導(dǎo)入速度大概在(300W/s左右)

　　2.4 insert into

　　insert into的方式主要應(yīng)用于日常關(guān)聯(lián)后的結(jié)果數(shù)據(jù)導(dǎo)入到新表的操作，為了測試insert into操作的速度以及影響，我們在一個時間段內(nèi)，連續(xù)大批量的導(dǎo)入到另外一張表來發(fā)現(xiàn)問題。最終發(fā)現(xiàn)，insert into的速度大概在780W/s，但是連續(xù)大批量的insert之后，大概連續(xù)導(dǎo)入了四批次，每次一百二十億的數(shù)據(jù)后(insert語句為：insert into tableA select*from tableB)，cpu一段時間內(nèi)占用會比較高，可能是內(nèi)部數(shù)據(jù)的合并操作導(dǎo)致的cpu使用上升。

　　三、數(shù)據(jù)查詢

　　下面主要選取了業(yè)務(wù)測試中比較重要的場景A作為測試，該測試主要測試日常事實和維度之間的關(guān)聯(lián)性能和面向業(yè)務(wù)的單表聚合查詢的性能。

　　3.1表模型選取

　　此次測試結(jié)合實際業(yè)務(wù)，我們主要測試的是明細(xì)表模型。DUPLICATE KEY選用的也是業(yè)務(wù)上常用來作為過濾條件的字段，采用的是按照天創(chuàng)建動態(tài)分區(qū)的方式建表，分布鍵根據(jù)業(yè)務(wù)特點的關(guān)系，基本上所有的表的分布鍵都是用一個字段。

　　3.2表創(chuàng)建方式

　　example:

　　create table ods.table1(

　　col1 datetime not null comment"time1",

　　col2 varchar(128)not null comment"str1",

　　col3 varchar(64)not null comment"str2",

　　col4 TINYINT not null comment"0,1,2",

　　col5 varchar(128)not null comment"str3",

　　col6 datetime not null comment"time2",

　　col7 TINYINT not null comment"-1,0,1,2,3"

　　)

　　DUPLICATE KEY(col1,col2,col3)

　　PARTITION BY RANGE(col1)(

　　PARTITION p20210101 values less THAN("2021-01-02 00:00:00"),

　　PARTITION p20210102 values less THAN("2021-01-03 00:00:00"),.

　　PARTITION p20210330 values less THAN("2021-03-31 00:00:00"),

　　PARTITION p20210331 values less THAN("2021-04-01 00:00:00")

　　)

　　DISTRIBUTED BY HASH(col3)BUCKETS 128

　　PROPERTIES(

　　"replication_num"="1",

　　"dynamic_partition.enable"="true",

　　"dynamic_partition.time_unit"="DAY",

　　"dynamic_partition.start"="-110",

　　"dynamic_partition.end"="2",

　　"dynamic_partition.prefix"="p",

　　"dynamic_partition.buckets"="128"

　　);

　　其余的表的創(chuàng)建方式類似，字段不同。

　　在后續(xù)的join過程中，由于預(yù)先沒有給表設(shè)置屬性Colocation Group，因此我們使用的alter方式修改每個表的Colocation Group屬性。如下：

　　ALTER TABLE table1 SET("colocate_with"="cg_col3");

　　3.3查詢測試

　　3.3.1關(guān)聯(lián)測試

　　1.hash join

　　默認(rèn)的join的方式為hash join，會使用JOIN(BROADCAST)的方式

　　2.shuffle join

　　在join的后邊顯示的指定[shuffle]的方式，會不采用廣播，而是用shuffle的方式進(jìn)行join。

　　如果某些情況下使用默認(rèn)的join時，右表數(shù)據(jù)量較大，廣播到多個be節(jié)點時會造成不可忽略的性能開銷，或者查詢直接oom導(dǎo)致be掛掉，可以嘗試使用此方式進(jìn)行查詢優(yōu)化。

　　3.colocation join

　　如果待關(guān)聯(lián)的兩張表的分布鍵和buckets數(shù)量一致，同時join的key是分布鍵，那么可以使用colocation join的方式進(jìn)行本地join。

　　由于數(shù)據(jù)會根據(jù)分布鍵進(jìn)行hash分布，相同分布鍵的數(shù)據(jù)處于同一個機器上，在join的時候數(shù)據(jù)只會在本地進(jìn)行join，避免跨網(wǎng)絡(luò)IO。

　　4.性能對比(全數(shù)據(jù)join之后的count)(大概的均值)

　　DorisDB：

　　左表數(shù)量：736億右表數(shù)量：15億

　　默認(rèn)的join：oom

　　shuffle join:90S

　　colocation join：60S

　　GP(極限是不到十億join不到一億，耗時近1800s)：

　　跑不動!!!

　　3.3.2單表查詢測試

　　DorisDB：

　　group by的字段為DUPLICATE KEY中的部分或者全部字段。單表數(shù)據(jù)量為736億。

　　邏輯為：全數(shù)據(jù)量下的select count(a.col1)as num from(select col1 as col1,col2 as col2 from table1 group by col1,col2)a;去重后的col1數(shù)據(jù)量為125億

　　耗時：600s

　　3.4參數(shù)優(yōu)化

　　以下參數(shù)在使用過程中需要根據(jù)實際情況進(jìn)行具體調(diào)整：

　　1.exec_mem_limit

　　該參數(shù)影響的地方很多，導(dǎo)入，查詢oom時可加大。建議可以設(shè)為機器內(nèi)存資源的70%-80%(只有doris進(jìn)程情況下)

　　2.is_report_success

　　該參數(shù)設(shè)為true后可以比較方便的查看物理執(zhí)行計劃

　　3.parallel_fragment_exec_instance_num

　　該參數(shù)影響查詢時的并行度，建議為機器core數(shù)的一半，查詢并發(fā)小的情況下可以酌情增加

　　4.query_timeout

　　查詢或者insert的超時時間，數(shù)據(jù)量大的情況下可以增加該參數(shù)

　　5.disable_storage_page_cache

　　在內(nèi)存資源充足的情況下，可以開啟page cache，啟用DorisDB自己維護(hù)的page cache，加速查詢

　　6.storage_page_cache_limit

　　開啟page cache占用的內(nèi)存大小，酌情設(shè)置。

　　在經(jīng)過測試后，DorisDB能夠滿足我司替換原有g(shù)reenplum集群，解決原有業(yè)務(wù)。

　　作者:劉志亮，就職于西安某安全公司，負(fù)責(zé)大數(shù)據(jù)相關(guān)內(nèi)容，專注于大數(shù)據(jù)方向研究