易仓跨境Saas全球租户，如何做到数据秒级响应？

Posted 2022-05-16 阿里云大数据AI技术

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了易仓跨境Saas全球租户，如何做到数据秒级响应？相关的知识，希望对你有一定的参考价值。

前言

易仓科技成立于2013年，致力于构建智能协同的跨境网络，让全球贸易更轻松。经过9年的积累沉淀，产业链SaaS+生态链协同的服务模式，已发展成为跨境电商行业的头部企业。目前租户分布全球，数据来自各大洲，汇集国外各大电商平台各个站点销售数据。面对这种数据体量大，跨全球各区域的复杂场景易仓大数据团队是如何做到秒级响应的？

问题 QUESTION

①数据分布三大洲，如何集中统一计算？

②Saas租户数据库级别物理隔离，分系统分库分表，百万张表如何同步？(1) A客户，A系统，两个库总共1000+张表与B客户，A系统同样的1000+张表
(2) Saas租户不同套餐保护不同功能系统，各系统表数量在100、200到500张不等
(3) A、B客户在同系统在同一个数据库实例与单客户单数据库实例（rds实例，polarDB实例）
(4) 几百个数据库实例分布在不同地域，不同国家

③如何在保证数据准确的基础上达到毫秒级别同步入仓？

技术选型

易仓跨境Saas全球租户，如何做到数据秒级响应？_数据库

针对以上问题的特殊性，复杂性，我们在技术选型上做了大量的调研以及深度测试，并与云服务团队深度交流做出了大量优化。

一、对于问题①②，数据分布以及多租户多库多表，如何处理？

易仓跨境Saas全球租户，如何做到数据秒级响应？_Flink_02

1.租户数据库分布在阿里云rds以及polarDB的几百个实例，分散在国内外各大Region，dts同时支持全库全表同步以及可选择库表。单个dts同步表数量没有限制，高度匹配易仓复杂的业务场景，在实施过程的功能以及性能优化是问题解决关键。

2.dts数据同步将国内外数据集中实时同步到统一华南区域的Kafka，保证数据准确，时效在毫秒级别。

二、数据准确毫秒级别同步入仓，关键点在Flink的高并发pipeline处理数据，时延毫秒级，且兼具可靠性

3.由Flink实时消费Kafka数写到实时数仓Hologres。

4.借助Maxcompute(下文简称MC)与实时数仓Hologres的互通能力，完成复杂的分析计算。

遇到的坑

1.DTS规格预估不准导致数据同步不稳定且没有及时的告警机制无法及时处理，无法保证数据准确性。

2.底层数仓建模不清晰，对于物理删除的客户数据数据中心采用逻辑删除进行标记后利用MC进行合并计算，以至于跟数据源无法一比一比对。无法判断数据同步的准确性。

3.以上导致了频繁的客户数据重推，数据初始化，以至于Kafka数据大量堆积Flink消费不及时。

4.以上步骤环环相扣，陷入数据异常同步的死循环，导致数据不可用。

如何解决

1.根据数据库实例实际RPS采用DTS对应的规格，推动DTS完善告警机制。数据同步稳定性实时性都大幅度提高。

(1) 推动DTS团队解决百万表级别数量的Api接口支持，开发互相支持快速迭代

(2) 定位大数据量表初始化任务频繁重启问题，优化dts任务元数据存储逻辑策略

易仓跨境Saas全球租户，如何做到数据秒级响应？_Saas_03

2.数据贴源层跟数据库一比一，重构底层设计方案，每个数据原表在原建表基础上通过sharing_seq、db_seq、company_code加上原表的主键组成唯一索引构成数据中心的目的表。

易仓跨境Saas全球租户，如何做到数据秒级响应？_Flink_04

-- 租户源表
CREATE TABLE `erp_test` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `d_key` varchar(50) NOT NULL DEFAULT  COMMENT 字典key,
  `d_val` varchar(100) NOT NULL DEFAULT  COMMENT 字典val,
  `d_desc` varchar(100) NOT NULL DEFAULT  COMMENT 字典描述,
  `ec_update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT 更新时间,
  `ec_create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 创建时
间,
  PRIMARY KEY (`id`),
  KEY `idxx_ec_update_time` (`ec_update_time`),
  KEY `idxx_ec_create_time` (`ec_create_time`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 COMMENT="erp_test表;


-- 数据中心目的表
CREATE TABLE `erp_test` (
  `sharing_seq` text,
  `db_seq` text,
  `company_code` text,
  `id` bigint NOT NULL,
  `d_desc` text,
  `d_key` text,
  `d_val` text,
  `ec_create_time` timestamptz,
  `ec_update_time` timestamptz
,PRIMARY KEY (`sharing_seq`, `db_seq`, `company_code`, `id`)
);

3.Flink核心代码逻辑优化，达到数据一比一还原数据库大量的新增、更新以及物理删除操作。数据可对比，有迹可循。通过大量数据对比，数据准确可靠并且实时性达到秒级

class DtsToDataworksHoloStream(fromSource: TSource[JSONObject], toSink: TSink[JSONObject]) extends TStream[JSONObject, JSONObject] 
  override val source: TSource[JSONObject] = fromSource
  override val sink: TSink[JSONObject] = toSink


  override protected def transform(dataStream: DataStream[JSONObject], others: DataStream[_]*): DataStream[JSONObject] = 
    val parallelism: Int = AppUtil.env.getParallelism
    val confStream: DataStream[mutable.Map[String, DbInfo2DataworksHoloBean]] = others.head.asInstanceOf[DataStream[mutable.Map[String, DbInfo2DataworksHoloBean]]]
    //获取各租户广播配置信息
    val confStreamDescriptor = new MapStateDescriptor[String, DbInfo2DataworksHoloBean]("DtsConfig", classOf[String], classOf[DbInfo2DataworksHoloBean])
    val confBroadcast: BroadcastStream[mutable.Map[String, DbInfo2DataworksHoloBean]] = confStream.broadcast(confStreamDescriptor)
    //Dts推送kafka的canalJson与租户广播配置connect
    val result: DataStream[JSONObject] = dataStream.connect(confBroadcast).process(new Dts2DataworksHoloTransformFunction)
    .keyBy(_.getString("table").hashCode % parallelism).map(s => s)
    result
  



//还原数据库增删改
def upsertOrDelTable(tableName: String, data: Array[(String, Any)], opType: String): Unit = 
    val put = new Put(holoclient.getTableSchema(tableName))
    if (opType == "DELETE") 
      put.getRecord.setType(SqlCommandType.DELETE)
    
    for (kv <- data) 
      put.setObject(kv._1.toLowerCase, kv._2)
    
    holoclient.put(put)