批量数据导入我们采用的是目前最主流的开源组件 DataX。自研的 DataOps 大数据管理平台在开源 DataX 的基础上做了很多封装，我们只需要创建数据同步任务，选择数据来源和数据目标表，即可自动生成字段映射和 DataX 规范的 Json 配置文件。
在项目初期，Doris 未发布 DataX 插件，仅通过原始的 JDBC 插入数据达不到性能要求。产品团队开发了 DataX 加速功能，先将对应数据抽取到本地文件，然后通过 Stream Load 方式加载入库，可以极大的提升数据抽取速度。数据读取到本地文件取决于网络宽带和本地读写性能，数据加载达到了 2000 千万数据 12.2G 仅需 5 分钟的效果。
此外，DataX 数据同步还支持读取自定义 SQL 的方式，通过自定义 SQL 可以处理 SQL SERVER 这种数据库比较难解决的字符转换问题和偶尔出现的乱码字符问题。
批量数据同步还支持增量模式，通过抽取最近 7 天的数据，配合 Doris 的主键模型，可以轻松解决大部分业务场景下的增量数据抽取。
在实时数据接入方面，由于接入的实时数据都来自于阿里云的 DRDS，所以我们采用的是 Canal+Kafka+Routine Load 模式。详细的配置就不展开了，环境搭建完成以后，只需要取 Canal 里面配置拦截策略，将表对应的流数据映射成 Kafka Topic，然后去 Doris 创建 R outine L oa d 就 OK 了，这里举一个 R outine L oad 的案例。
ALTER TABLE DS_ORDER_INFO ENABLE FEATURE "BATCH_DELETE"；CREATE ROUTINE LOAD t02_e3_zy.ds_order_info ON DS_ORDER_INFOWITH MERGECOLUMNS(order_id, order_sn, deal_code, ori_deal_code, ori_order_sn,crt_time = now(), cdc_op),DELETE ON cdc_op="DELETE" PROPERTIES ( "desired_concurrent_number"="1", "max_batch_interval" = "20", "max_batch_rows" = "200000", "max_batch_size" = "104857600", "strict_mode" = "false", "strip_outer_array" = "true", "format" = "json", "json_root" = "$.data", "jsonpaths" = "[\"$.order_id\",\"$.order_sn\",\"$.deal_code\",\"$.ori_deal_code\",\"$.ori_order_sn\",\"$.type\" ]" )FROM KAFKA ( "kafka_broker_list" = "192.168.87.107:9092,192.168.87.108:9092,192.168.87.109:9092", "kafka_topic" = "e3_order_info", "kafka_partitions" = "0", "kafka_offsets" = "OFFSET_BEGINNING", "property.group.id" = "ds_order_info", "property.client.id" = "doris");
本次项目的数据加工我们是通过 Doris 提供的视图来完成的。利用 Doris 优秀的索引能力，加上完善的 SQL 语法支持，即使再复杂的逻辑，也可以通过视图来实现。用视图加工数据，减少了代码发布的流程，避免了编译错误的问题 比 Hive 的脚本式开发更加高效。
在完成模型设计以后，我们会确定模型表的命名、Key 类型等信息。完成表的创建以后，我们会创建表名 +"_v" 的视图，用于处理该表数据的逻辑加工。在大多数情况下，我们都是先清空目标表，然后从视图读取数据写入目标表的，所以我们的调度任务都比较简单，例如：
truncate table xtep_dw.dim_shop_info;insert into xtep_dw.dim_shop_infoselect * from xtep_dw.dim_shop_info_v;
对于数据量特别大的读写任务，则需要分步写入。例如：
truncate table xtep_dw.dwd_god_allocation_detail_drp;
insert into xtep_dw.dwd_god_allocation_detail_drpselect * from xtep_dw.dwd_god_allocation_detail_drp_v where UPDATE_DATE BETWEEN '2020-01-01' and '2020-06-30'; insert into xtep_dw.dwd_god_allocation_detail_drpselect * from xtep_dw.dwd_god_allocation_detail_drp_v where UPDATE_DATE BETWEEN '2020-07-01' and '2020-12-31';
insert into xtep_dw.dwd_god_allocation_detail_drpselect * from xtep_dw.dwd_god_allocation_detail_drp_v where UPDATE_DATE BETWEEN '2021-01-01' and '2021-06-30'; insert into xtep_dw.dwd_god_allocation_detail_drpselect * from xtep_dw.dwd_god_allocation_detail_drp_v where UPDATE_DATE BETWEEN '2021-07-01' and '2021-12-31';
insert into xtep_dw.dwd_god_allocation_detail_drpselect * from xtep_dw.dwd_god_allocation_detail_drp_v where UPDATE_DATE BETWEEN '2022-01-01' and '2022-06-30'; insert into xtep_dw.dwd_god_allocation_detail_drpselect * from xtep_dw.dwd_god_allocation_detail_drp_v where UPDATE_DATE BETWEEN '2022-07-01' and '2022-12-31';
当然，视图开发的模式也是有一定的弊端的，比如不能做版本管理，也不便于备份。为此，我们承受了一次惨痛教训，在项目测试阶段，有同事在 dbwaver 客户端误操作 Drop 掉了 xtep_dw 数据库，导致我们花费了 3-4 天时间才恢复程序。 因此我们紧急开发了 Python 备份程序：
import pymysqlimport json import sys import osimport timeif sys.getdefaultencoding() != 'utf-8': reload(sys) sys.setdefaultencoding('utf-8')
BACKUP_DIR='/data/cron_shell/database_backup'BD_LIST=['ods_add','ods_cdp','ods_drp','ods_e3_fx','ods_e3_jv','ods_e3_pld','ods_e3_rds1','ods_e3_rds2','ods_e3_zy','ods_hana','ods_rpa','ods_temp','ods_xgs','ods_xt1','t01_hana','xtep_cfg','xtep_dm','xtep_dw','xtep_fr','xtep_rpa']
if __name__ == "__main__": basepath = os.path.join(BACKUP_DIR,time.strftime("%Y%m%d-%H%M%S", time.localtime())) print basepath if not os.path.exists(basepath): os.makedirs(basepath) conn = pymysql.connect( host='192.168.xx.xx', port=9030, user='root', password='****', database='information_schema', charset='utf8')
cursor = conn.cursor() for dbname in BD_LIST: dbpath = os.path.join(basepath ,dbname) print(dbpath) if not os.path.exists(dbpath): os.makedirs(dbpath) sql1 = "select TABLE_SCHEMA,TABLE_NAME,TABLE_TYPE from information_schema.tables where TABLE_SCHEMA ='%s';"%(dbname) print(sql1) cursor.execute(sql1) for row in cursor.fetchall(): tbname = row[1] filepath = os.path.join(dbpath ,tbname + '.sql') print(u'%s表结构将备份到路径：%s'%(tbname,filepath)) sql2 = 'show create table %s.%s'%(dbname,tbname) print(sql2) cursor.execute(sql2) for row in cursor.fetchall(): create_sql = row[1].encode('GB18030') with open(filepath, 'w') as fp: fp.write(create_sql) cursor.close() conn.close()
然后配合 Linux 的 Crontab 定时任务，每天三次备份代码。
0 8,12,18 * * * python /data/cron_shell/backup_doris_schema.py >>/data/cron_shell/logs/backup_doris_schema.log
本次项目采用的前端工具是某国产 BI 软件和定制化开发的 E-Charts 大屏。该 BI 软件是基于数据集为中心去构建报表，并且支持灵活的数据权限管理。大多数据情况下我们基于SQL查询创建数据集，可以有效过滤数据。本次项目基于该 BI 平台构建了 PC 端报表（通过 App 适配手机也可以直接访问），并且新增了自助分析报表，同步开发了几张 E-Charts 大屏，实现大屏、中屏、小屏的统一。
数据查询对 Doris 来说是很 Easy 的了，基本上建好表以后设置好索引，利用好 B itmap ，性能就不会差。 这里需要说明的上，在性能压力不大的情况下合理使用视图来关联多个结果集，可以减少跑批的任务和数据处理层级，有利于报表数据的快速刷新。 在这方面，我们也是尽可能减少 DWS 和 ADS 层的聚合模型，减少大数据量的读写，尽可能复用代码逻辑和模型表，减少跑批时间加强系统稳定性。
在 实时的需求方面，我们分别尝试了 Lambda 架构和 Kappa 架构，最后走出来项目特色的第三条线路——相同的视图逻辑，用不同的调度任务刷新不同范围的数据，实现流批代码复用。
项目早期，我们是按照 Lambda 架构构建的任务，系统数据分为批处理和流处理两条线路，随着批处理的稳定，流处理数据的不准确性就逐步暴露。究其原因，业务系统存在数据物理删除和更新的情况，双流 Join 之后的数据准确性得不到保障。再有就是同时维护两套代码，实时逻辑的更新滞后，变更逻辑的代价也比较大。
项目后期，基于业务要求我们也尝试了把所有的零售逻辑搬迁到流处理平台，以实现流批一体，但是发现无法处理报表上常规要求本同期对比、维度数据变更和复杂条件过滤，导致搬迁工作半途而废。
最后我们结合项目的实际情况，采用批处理和微批处理结合的方式，一套代码，两种跑批模式。T+1 的链路执行最近 6 个月或者全量数据的刷新，微批处理流程刷新当日数据或者本周数据，实现数据的快速迭代更新。以 DWB 层为例：
truncate table xtep_dw.dwb_ret_sales;insert into xtep_dw.dwb_ret_sales select * from xtep_dw.dwb_ret_sales_v;delete from xtep_dw.dwb_ret_sales where report_date='${curdate}';insert into xtep_dw.dwb_ret_sales select * from xtep_dw.dwb_ret_sales_vwhere report_date='${curdate}';
而 DWS 和 ADS 层的情况更为复杂，由于跑批频率太高，为了避免出现用户查看报表时刚好数据被删除的情况，我们采用分区替换的方式来实现数据的无缝切换。
truncate table xtep_dw.dws_ret_sales_xt_swap;insert into xtep_dw.dws_ret_sales_xt_swapselect * from xtep_dw.dws_ret_sales_xt_vwhere date_tag in ('本日','本周','本月','本年');insert into xtep_dw.dws_ret_sales_xt_swapselect * from xtep_dw.dws_ret_sales_xt_vwhere date_tag in ('昨日','上周','上月','上年');ALTER TABLE xtep_dw.dws_ret_sales_xt REPLACE WITH TABLE dws_ret_sales_xt_swap;truncate table xtep_dw.dws_ret_sales_xt_swap; insert into xtep_dw.dws_ret_sales_xt_swapselect * from xtep_dw.dws_ret_sales_xt_v where date_tag in ('本日','本周');ALTER TABLE xtep_dw.dws_ret_sales_xt ADD TEMPORARY PARTITION tp_curr1 VALUES IN ('本日','本周');insert into xtep_dw.dws_ret_sales_xt TEMPORARY PARTITION (tp_curr1) SELECT * from xtep_dw.dws_ret_sales_xt_swap;ALTER TABLE xtep_dw.dws_ret_sales_xt REPLACE PARTITION (p_curr1) WITH TEMPORARY PARTITION (tp_curr1);
弱弱的说一 下，Doris 的分区替换还需要再完善一下，希望可以支持类似于 ClickHouse 的语法，即：ALTER TABLE table2 REPLACE PARTITION partition_expr FROM table1;
同时，由于我们设计了良好的分层架构，对于实时性要求特别高的数据，例如双十一大屏，我们可以直接从 ODS 层汇总数据到报表层，可以实现秒级的实时查询；对于实时性较高的业务，例如移动端实时日报，我们从 DWD 或者 DWB 往上汇总数据，可以实现分钟级的实时；对于普通的自助分析或者固定报表，则按照灵活的频率更新数据，兼顾了二者的时效性和准确性。
在项目过程中，我们还遇到一些其它问题，这里简单总结一下。
查询任务耗用的内存过大，导致 Doris BE 挂掉的情况，我们也出现过。我们采取的方法是所有表都创建 3 副本，然后给 Doris 进程配置 Supervisord 自启动进程，失败的任务通过调度平台的重试功能，一般都可以在 3 次重试机会以内跑过。
批处理过程中确实会有一些复杂的任务或者写入数据太多的任务会超时，除了调大 timeout 参数（目前设置为 10 分钟）以外，我们还把任务做了切分。前面的调度任务案例已经可以看到，有些写入的 SQL 我们是按照分区字段或者日期区间来分批计算和写入的。
删除语句不支持表达式，我认为是 Doris 后续需要优化的一个功能点。在 Doris 无法实现的情况下，我们通过改造调度平台的参数功能，先计算好参数值，然后传入变量的方式实现了动态条件删除数据。前文的调度任务代码也有案例。
误删除重要的表是数据仓库开发过程中比较常见的情况，表结构我们可以通过 Python 做好备份，但是表数据实在没有更好的办法。这里 Doris 提供了一个很好的功能——Recover 功能，推荐给大家。误删除的表在 1 天以内可以支持闪回。
目前 Apache Doris 在特步集团的应用已经得到了用户的认可，今年 2 月底又对 Doris 集群进行了硬件升级，接下来会基于现有的接口数据拓展到特步品牌 BI 应用，并且迁移更多的 HANA 数仓应用到 Doris 平台。随着应用的深入，我们需要加强对 Doris 集群、Routine Load 和 Flink 任务的监控，及时发出异常预警，缩短故障恢复时间。同时，随着向量化引擎的逐步成熟和查询优化器的进一步完善，我们需要调整一些 SQL 写法，降低批处理对系统资源的占用，让集群更好的同时服务批处理和查询需求。当然，也期待社区在资源隔 离方面可以有更进一步的完善。
最后提一个重要的产品优化方向，希望社区予以考虑： 为了可以更好的用 Doris 替代 Hive 数仓，希望社区可以考虑开发存储过程功能。
最后，感谢 Apache Doris 社区给予的支持，也感谢百度开源了这么优秀的产品！同时要感谢信息部领导林总和曾经理给予的大力支持与包容。祝愿 Apache Doris 社区发展越 来越好！
- 作者介绍 -
杨宏武
特步零售数据仓库项目架构师，Apache Doris 活跃 Contributor，负责公司产品研发和 Apache Doris 应用方向的落地和推广。
王春波
特步零售数据仓库项目技术经理，《高效使用Greenplum：入门、进阶与数据中台》的作者，“数据中台研习社”号主，零售数仓项目实施专家。
