原Greenplum集群在进行10亿*1亿的关联查询时达到极限，无法支撑更大数据量级的关联查询。使用DorisDB替换Greenplum构建新的集群，在进行736亿*15亿的超大量级数据关联查询时，不仅可以顺利完成，并且耗时很短，对业务的整体提升巨大。
一、使用背景
1.1 选用原因
我司原有业务查询使用的数据库为Greenplum，在数据源变更后，数据量从原来的日增千万级别（近百G）暴增至日增千亿（10T）级别，原有的12台GP集群在数据量增长后存在以下痛点：
- 1、数据导入
原有的数据导入借助于gpload的工具，在有索引的情况下，数据导入随着数据量的增加会变慢，在千亿级日增情况下，有索引的表根本无法导入。即使使用先导入数据，后建索引的方式，导入过程还是不理想，建索引的时间会由于数据量的增长而增长，由于机器资源在现有的基础上增加的的可能性不是很大，使用该方式做数据导入，整个流程耗时相当长，无法满足业务需求。
- 2、数据存储
GP在数据存储这一块，如果使用heap表的方式创建表，数据来说是不做任何压缩进行存储，比较占用存储资源。如果采用列存表的方式，需要手动指定压缩等级和字段，但是使用者在不清楚数据重复的具体情况下设置该参数就只能是想当然去做，然后在生产中根据数据实际情况进行更改，在查询时，cpu会进行解压缩操作，增加了cpu的计算耗时。
- 3、数据计算
计算瓶颈其实是我们数据量增长之后主要的痛点。在原有的使用过程中，针对于业务A的整体运行时长，从客户触发到最终显示，需要大概100分钟左右，数据量增长后，业务A的基本跑不动。其次，在日常的etl过程中，一些定时表关联在原GP的处理过程中只能是对事实表按照时间粒度做切分，小部分小部分的进行关联，然后再进行合并处理，数据量增长之后的关联，在现有资源下也无法实现。
在GP无法承受如此巨大的数据量，满足不了业务的需求时，我们将目光转向其他解决方案，在测试了DorisDB，clickhouse以及其他olap产品后，结合自身的业务特点和使用上的易用性，最终选用了DorisDB作为MPP的解决方案。此文档也是基于DorisDB进行详细的业务测试过程中整理的文档。
1.2 集群配置
此次测试使用的机器资源如下所示(只部署了DorisDB的环境)：
此次测试使用的机器资源如下所示(只部署了DorisDB的环境)：
机器数量：10台
机器系统：centos7.6
机器内存：256G
机器磁盘：7200转机械硬盘，每台机器为8T*4,做了raid0
网络带宽：内网万兆光迁
CPU: 2*12 core
此次部署的DorisDB的集群详情如下（未使用spark load，没有安装spark的客户端）：
fe数量：3台（1 master + 2 follower）
be数量：10台
broker数量：10台
1.3 集群配置参数
针对自身业务特点，修改了以下参数：
fe:
broker load的参数
1.允许运行的最大的broker数量
max_broker_concurrency=10
2.每个be处理的数据量
max_bytes_per_broker_scanner=32212254720
上述两个参数影响broker load导入时be处理数据的并发数量和单个be处理的数据量
文件合并的参数
1.be节点base compaction线程数量
base_compaction_num_threads_per_disk=4
2.base compaction时写磁盘的限速，单位为M
base_compaction_write_mbytes_per_sec=20
3.be节点cumulative compaction的线程数量
cumulative_compaction_num_threads_per_disk=8
4.be节点cumulative compactiond写磁盘的限速，单位为M
cumulative_compaction_write_mbytes_per_sec=300
5.be节点cumulative compactiond线程轮询的间隔
cumulative_compaction_check_interval_seconds=2
上述五个参数主要控制DorisDB对于文件合并的效率，可以根据自身的硬件性能和实际业务情况调整该参数。大量数据导入到DorisDB中时，DorisDB需要根据排序key做排序，根据字段的值做压缩合并的操作，此时会占用磁盘性能，调整该参数（业务闲时）可以加速这一过程，使DorisDB专注于计算。
以上参数仅提供参考，请根据自身资源和实际情况酌情调整
二、数据导入
由于数据源的特殊性，数据存放在文件中，原始文件为压缩文件，因此在实际测试过程中，我们主要对以下几种导入进行了测试（spark load未测试成功），最终选取了broker load的方式作为最终的数据导入的方案。该方案能够实现单任务200W+/s的导入速度，并且支持并行的方式，进一步提高数据导入速度。
2.1 stream load
刚开始使用DorisDB时，我们使用的导入方式即为stream load的方式测试小批量的数据，但是在数据量增大的情况下，大概数据单次导入到100G时，发现这种微批导入的方式有数据膨胀的情况，导入前的数据和入库后的数据量对比差异明显（导入前100G左右，导入后DorisDB在250G左右，并且磁盘IO占用高居不下），遂放弃。
2.2 datax
datax主要是由于有丰富的使用经验，其次是datax在对于数据接入过程中很灵活，可以增加很多丰富的transformer插件来减轻后续的数据清洗的压力。datax使用时我们主要使用的是mysqlWriter和利用stream load实现的DorisWriter（此处艾特社区张怀北同学）。前者在我们测试时，DorisDB文档中还未增加doriswriter的内容，利用的是mysql的jdbc连接实现数据的导入。后者则是社区利用stream load 的api实现的数据导入。在使用过程中，发现导入速度并不理想（10台一起跑，导入速度在60W/s），满足不了我们每天的增量数据的导入的要求。也放弃datax的导入方案。（如果有对这种方式感兴趣的同学可以在社区留言）。文章末尾附件有该writer实现的核心代码。
2.3 broker load
broker load 的导入方式是我们最终采用的方案。原本对spark load的方式抱有很大希望，因为我们业务中的数据另一个导入方向为hbase，使用的导入方式为bulkload的方式，利用spark 合成Hfile的方式写入hbase，该方式能够将待导入的数据进行排序后，形成hbase底层需要的hfile的格式写入到hdfs，hbase可以不用再将数据在内存中排序后再落盘，在进行合并形成hfile，能够借助于spark计算集群减轻hbase排序和文件合并的压力，使得hbase专注于业务。我们猜想DorisDB的spark load是否也采用了类似的思想，利用spark处理数据后直接生成DorisDB所需要的底层存储文件后写入DorisDB，但是在经过咨询后，现有的spark load不具备这种提前排序生成底层存储文件的导入功能，但是在未来会开发。后续开发完成后，对于DorisDB的导入应该提升很大（个人臆想_）。
broker load的时候我们测试了分别从hdfs load csv文件和parquet文件，最终发现使用parquet导入比csv性能高出两倍到三倍的样子（相同数据条数，字段），也刚好是parquet文件和csv文件实际存储相差的样子。同时在导入时，可以先将待导入的表的副本设为1，可以减少导入过程中的数据clone，加快导入速度。最终测的的导入速度大概在（300W/s左右）
2.4 insert into
insert into的方式主要应用于日常关联后的结果数据导入到新表的操作，为了测试insert into操作的速度以及影响，我们在一个时间段内，连续大批量的导入到另外一张表来发现问题。最终发现，insert into的速度大概在780W/s，但是连续大批量的insert之后，大概连续导入了四批次，每次一百二十亿的数据后(insert语句为：insert into tableA select * from tableB)，cpu一段时间内占用会比较高，可能是内部数据的合并操作导致的cpu使用上升。
三、数据查询
下面主要选取了业务测试中比较重要的场景A作为测试，该测试主要测试日常事实和维度之间的关联性能和面向业务的单表聚合查询的性能。
3.1 表模型选取
此次测试结合实际业务，我们主要测试的是明细表模型。DUPLICATE KEY选用的也是业务上常用来作为过滤条件的字段，采用的是按照天创建动态分区的方式建表，分布键根据业务特点的关系，基本上所有的表的分布键都是用一个字段。
3.2 表创建方式
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"
);
其余的表的创建方式类似，字段不同。
在后续的join过程中，由于预先没有给表设置属性Colocation Group，因此我们使用的alter方式修改每个表的Colocation Group属性。如下：
ALTER TABLE table1 SET ("colocate_with" = "cg_col3");3.3 查询测试
3.3.1 关联测试
1. hash join
默认的join的方式为hash join，会使用JOIN (BROADCAST)的方式
2. shuffle join
在join的后边显示的指定 [shuffle]的方式，会不采用广播，而是用shuffle的方式进行join。
如果某些情况下使用默认的join时，右表数据量较大，广播到多个be节点时会造成不可忽略的性能开销，或者查询直接oom导致be挂掉，可以尝试使用此方式进行查询优化。
3. colocation join
如果待关联的两张表的分布键和buckets数量一致，同时join的key是分布键，那么可以使用colocation join的方式进行本地join。
由于数据会根据分布键进行hash分布，相同分布键的数据处于同一个机器上，在join的时候数据只会在本地进行join，避免跨网络IO。
4. 性能对比(全数据join之后的count)(大概的均值)
DorisDB：
左表数量：736亿 右表数量：15亿
默认的join ：oom
shuffle join: 90S
colocation join：60S
GP(极限是不到十亿join不到一亿，耗时近1800s)：
跑不动!!!
3.3.2 单表查询测试
DorisDB：
group by 的字段为DUPLICATE KEY中的部分或者全部字段。单表数据量为736亿。
逻辑为： 全数据量下的select count(a.col1) as num from (select col1 as col1,col2 as col2 from table1 group by col1,col2)a; 去重后的col1数据量为125亿
耗时：600s
3.4参数优化
以下参数在使用过程中需要根据实际情况进行具体调整：
1.exec_mem_limit
该参数影响的地方很多，导入，查询oom时可加大。建议可以设为机器内存资源的70%-80%（只有doris进程情况下）
2.is_report_success
该参数设为true后可以比较方便的查看物理执行计划
3.parallel_fragment_exec_instance_num
该参数影响查询时的并行度，建议为机器core数的一半，查询并发小的情况下可以酌情增加
4.query_timeout
查询或者insert的超时时间，数据量大的情况下可以增加该参数
5.disable_storage_page_cache
在内存资源充足的情况下，可以开启page cache，启用DorisDB自己维护的page cache，加速查询
6.storage_page_cache_limit
开启page cache占用的内存大小，酌情设置。
在经过测试后，DorisDB能够满足我司替换原有greenplum集群，解决原有业务。
________________________________________
此文档只是我司针对自己业务进行的部分测试的一个文档，欢迎大佬指出文档中不正确的地方，互相交流，共同进步
附件
利用stream load实现的writer核心task代码
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package com.alibaba.datax.plugin.writer.doriswriter;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.codec.binary.Base64;
import org.apache.http.HttpHeaders;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.UUID;
public class DorisWriterTask
{
private static final Logger LOG = LoggerFactory.getLogger(DorisWriterTask.class);
private final Configuration configuration;
private String username;
private String password;
private String loadUrl;
private List<Object> heads;
private List<String> column;
private static final String SEPARATOR = "|";
private int batchSize;
private HttpClientBuilder httpClientBuilder;
public DorisWriterTask(Configuration configuration) {this.configuration = configuration;}
public void init()
{
List<Object> connList = configuration.getList(Key.CONNECTION);
Configuration conn = Configuration.from(connList.get(0).toString());
String endpoint = conn.getString(Key.ENDPOINT);
String table = conn.getString(Key.TABLE);
String database = conn.getString(Key.DATABASE);
this.column = configuration.getList(Key.COLUMN, String.class);
// 如果 column 填写的是 * ，直接设置为null，方便后续判断
if (this.column!= null && this.column.size() == 1 && "*".equals(this.column.get(0))) {
this.column = null;
}
this.batchSize = configuration.getInt(Key.BATCH_SIZE, 1024);
this.heads = configuration.getList(Key.HEADS);
this.username = configuration.getString(Key.USERNAME);
this.password = configuration.getString(Key.PASSWORD, null);
this.loadUrl = String.format("%s/api/%s/%s/_stream_load", endpoint, database, table);
this.httpClientBuilder = HttpClients
.custom()
.setRedirectStrategy(new DefaultRedirectStrategy()
{
@Override
protected boolean isRedirectable(String method)
{
return true;
}
});
LOG.info("connect DorisDB with {}", this.loadUrl);
}
private String basicAuthHeader(String username, String password)
{
String tokenEncode = username + ":" + password;
byte[] encoded = Base64.encodeBase64(tokenEncode.getBytes(StandardCharsets.UTF_8));
return "Basic " + new String(encoded);
}
public void startWrite(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector)
{
Record record;
int currSize = 0;
StringBuilder stringBuilder = new StringBuilder();
while ((record = recordReceiver.getFromReader()) != null) {
int len = record.getColumnNumber();
if (this.column != null && len != this.column.size()) {
throw DataXException.asDataXException(
DorisWriterErrorCode.ILLEGAL_VALUE,
String.format("源字段数和目标字段数不匹配，源字段数为%s, 目标字段数为%s", len, this.column.size())
);
}
StringBuilder oneRow = new StringBuilder();
for (int i = 0; i < len; i++) {
if (record.getColumn(i).getRawData() != null) {
oneRow.append(record.getColumn(i).asString());
}
if (i < len - 1) {
oneRow.append(SEPARATOR);
}
}
oneRow.append("\n");
stringBuilder.append(oneRow);
currSize++;
if (currSize >= this.batchSize) {
stringBuilder.deleteCharAt(stringBuilder.length() - 1);
sendData(stringBuilder.toString());
currSize = 0;
stringBuilder.setLength(0);
}
}
if (stringBuilder.length() > 0) {
sendData(stringBuilder.toString());
}
}
private void sendData(String content)
{
try (CloseableHttpClient client = this.httpClientBuilder.build()) {
HttpPut put = new HttpPut(this.loadUrl);
StringEntity entity = new StringEntity(content, "UTF-8");
put.setHeader(HttpHeaders.EXPECT, "100-continue");
if (this.username != null && this.password != null) {
put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(this.username, this.password));
}
put.setHeader("column_separator", SEPARATOR);
// the label header is optional, not necessary
// use label header can ensure at most once semantics
put.setHeader("label", UUID.randomUUID().toString());
if (this.column != null ) {
put.setHeader("columns", String.join(",", this.column));
}
if (this.heads != null && this.heads != null) {
for (int i = 0; i < this.heads.size(); i++) {
System.out.println(this.heads.get(i));
JSONObject head = (JSONObject) this.heads.get(i);
for(Map.Entry<String,Object> entry : head.entrySet()){
String key = entry.getKey();
Object value = entry.getValue();
System.out.println("key:::"+key);
System.out.println("value:::"+value.toString());
put.setHeader(key, value.toString());
}
}
}
put.setEntity(entity);
try (CloseableHttpResponse response = client.execute(put)) {
String loadResult = "";
if (response.getEntity() != null) {
loadResult = EntityUtils.toString(response.getEntity());
}
int statusCode = response.getStatusLine().getStatusCode();
// statusCode 200 just indicates that doris be service is ok, not stream load
// you should see the output content to find whether stream load is success
if (statusCode != 200) {
throw DataXException.asDataXException(
DorisWriterErrorCode.WRITER_ERROR,
String.format("Stream load failed, statusCode=%s load result=%s", statusCode, loadResult)
);
}
}
}
catch (IOException e) {
throw DataXException.asDataXException(
DorisWriterErrorCode.CONNECT_ERROR,
String.format("Failed to connect Doris server with: %s, %s", this.loadUrl, e)
);
}
}
}