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

-- 租户源表
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`)
 );

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)  
  }