今天是Doris的 Contributor徐小冰 同学代表 搜狐 带来的关于Apache Doris (incubating)Parquet文件读取的设计与实现。
所有需求的推动都基于真实的业务痛点。搜狐在Hadoop上的文件主要存储为Parquet。
Parquet有如下优势：
- 列式存储，压缩比高(RLE、字段编码等)，查询效率高(列pruning，block filter)
- Spark/Impala/Hive都支持(ORC Impala最新版本才支持)
而Doris只支持CSV格式。因此Parquet文件的读取流程就需要两步：
- 通过相关命令行或者工具将数据表导出到csv文件中
- 通过Doris load命令进行导入
这种方式的问题在于一方面CSV默认换行符是\n,如果数据中包含该\n会导致load失败。并且整体效率不高。 因此，Doris支持Parquet读取，势在必行 。
Load流程概览
首先来了解一下load流程。Doris的整体架构如图所示。
首先通过MySQL Client和MySQL协议，将load请求下发到FE master上。MySQL协议是异步的，执行完成后就会直接返回给MySQL client。
如果导入多个文件，可能会需要在FE进行负载均衡，即分发给其他的FE（FE follower），FE会从BE中选择来处理，BE会分别通过broker读取HDFS，由BE进行处理。
Load流程
Load-Prepare流程
Load-Open流程
Load-Scanner流程
使用方式
显示指定文件格式为: parquet ，指定语句为：
FORMAT AS "parquet"
如若缺省则按照导入文件后缀名判断。
具体示例：
1 LOAD LABEL mydatabase.load_test1
2 (
3 DATA INFILE("hdfs://127.0.0.1:8020/tmp/xuxb/000444abc_0")
4
5 INTO TABLE parquet_table
6
7 FORMAT AS "parquet"
8
9 (col_a, col_b, col_c)
10 )
11
12 WITH BROKER broker_parquet;
同时也支持指定目录，指定语句同样为：
FORMAT AS "parquet"
通过 DATA INFILE 后的 /* 来指定目录文件
DATA INFILE(“hdfs://127.0.0.1:8020/tmp/xuxb/abc/*”)
具体示例：
1 LOAD LABEL mydatabase.load_test1
2
3 (
4
5 DATA INFILE("hdfs://127.0.0.1:8020/tmp/xuxb/abc/*")
6
7 INTO TABLE parquet_table
8
9 FORMAT AS "parquet"
10
11 (col_a, col_b, col_c)
12
13 )
14
15 WITH BROKER broker_parquet;
对于分区字段，在Parquet文件中是不存在的，在这种情况下应该如何应用？
例如id,type是两个分区字段，在Parquet文件中不存在该列信息， 首先创建Doris表如下，将id,type存在Doris表中 ：
1 CREATE TABLE demo_mem_infos
2
3 (
4
5 id INT, #分区字段 存在Doris表中
6
7 type INT, #分区字段 存在Doris表中
8
9 col_a VARCHAR(128),
10
11 col_b VARCHAR(128),
12
13 col_c VARCHAR(128)
14
15 )
16
17 ENGINE=olap
18
19 AGGREGATE KEY(`col_a`,`col_b`,`col_c`)
20
21 DISTRIBUTED BY HASH(id) BUCKETS 25
22
23 PROPERTIES ("storage_type"="column","replication_num" = "1");
Load的写法如下：
1 LOAD LABEL mydatabase.load_test1
2
3 (
4
5 DATA INFILE("hdfs://127.0.0.1:8020/tmp/id=100/type=10/*")
6
7 INTO TABLE demo_mem_infos
8
9 FORMAT AS "parquet"
10
11 (col_a,col_b,col_c)
12
13 set (
14
15 id=default_value("100"),
16
17 type=default_value("10")
18
19 )
20
21 )
22
23 WITH BROKER broker_parquet;
其中，在DATA INFILE中进行说明：
DATA INFILE(“hdfs://127.0.0.1:8020/tmp/id=100/type=10/*”)
并指定导入方式：
FORMAT AS "parquet"
并用set语句对这两个分区字段进行默认值的设置
set (
id=default_value(“100”),
type=default_value("1 0")
)
即可完成导入。
在导入过程中,通过show load查看状态, 因为Parquet压缩比比较高，可能会出现内存溢出问题:
报错：
type:ETL_RUN_FAIL; msg:Broker etl failed: Memory limit exceeded
可在执行load之前执行如下命令:
set exec_mem_limit = 64424509440 ; #单位为字节, 可根据实际情况进行设置。
可参考:FAQ
https://github.com/apache/incubator-doris/wiki/Doris-FAQ
Arrow库的引入
在实现读取Parquet文件的过程中引入了第三方库Arrow：
代码地址: https://github.com/apache/arrow
引入Arrow库的原因有几点:
- Arrow接口抽象度非常高,只需要写少量代码就可读取parquet文件
- 开发周期短,不用自己维护相关数据结构以及内存
具体的代码展示如下：
对现有的scanner对象进行抽象,定义抽象类BaseScanner。
读取RowGroup效率提升的实现
在开发中总共有两个版本，第一个版本效率非常低。于是深入研究了Parquet文件的格式。Parquet由多个RowGroup组成, 优化后每次读取一个完整RowGroup，大大提升了效率。
在实现上，先去调取文件含有多少个RowGroup，然后循环遍历，一次只读一个，提高了效率。
测试结果
导入27G的一张表,测试结果如下：
存在问题
-
内存过高
主要原因：由于Doris要做排序相关操作,目前版本(0.10.0)实现是把所有数据读到内存中,在做快排。
后续解决方案排序算法改为外排序。
-
varchar类型最大支持65535。如果存储一篇文章内容可能无法满足。
性能优化
-
Arrow接口不熟悉，第一版功能导入2G数据花费2个小时没有完成。通过BE log中profile信息发现网络io性能不佳，重新阅读Arrow源码+Arrow ut, 将实现方式改为每次读取一个完整RowGroup(主要优化点)
-
第二版完成后,性能任然不达标,通过ut+perf+火焰图,定位发现map find为性能瓶颈点,优化代码后显著提升。
代码贡献
-
将libarrow等相关第三方库编译加入到thirdparty中：
-
将相关实现提交be/fe/fs_broker中
-
解决Doris两个小问题
内存泄露问题
https://github.com/apache/incubator-doris/pull/1244
Mysql8.0连接Fe失败