前言

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

技术选型

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

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

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

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

遇到的坑

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

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

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

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

如何解决

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

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

-- 租户源表


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)

  }

亿万数据量级的业务订单表，实现了从数据源端到数据中心的业务延迟控制在毫秒级别，基于此数据中心秒级别的业务响应更是不在话下。

后 记

注：

① Flink ：一个批处理和流处理结合的统一计算框架

② Hologres ：一站式实时数仓引擎（ Real-Time Data Warehouse ）

③ Maxcompute ：大数据计算服务，快速、完全托管的 PB 级数据仓库解决方案