日均百億級日誌處理：微博基於Flink的實時計算平台建設

• 2019 年 11 月 15 日
• 筆記

作者介紹

呂永衛，微博廣告資深數據開發工程師，實時數據項目組負責人。

黃鵬，微博廣告實時數據開發工程師，負責法拉第實驗平台數據開發、實時數據關聯平台、實時演算法特徵數據計算、實時數據倉庫、實時數據清洗組件開發工作。

林發明，微博廣告資深數據開發工程師，負責演算法實時特徵數據計算、實時數據關聯平台、實時數據倉庫、FlinkStream組件開發工作。

崔澤峰，微博廣告資深數據開發工程師，負責實時演算法特徵數據計算、實時任務管理平台、FlinkStream組件、FlinkSQL擴展開發工作。

引言

是隨著微博業務線的快速擴張，微博廣告各類業務日誌的數量也隨之急劇增長。傳統基於Hadoop生態的離線數據存儲計算方案已在業界形成統一的默契，但受制於離線計算的時效性制約，越來越多的數據應用場景已從離線轉為實時。微博廣告實時數據平台以此為背景進行設計與構建，目前該系統已支援日均處理日誌數量超過百億，接入產品線、業務日誌類型若干。

一、技術選型

相比於Spark，目前Spark的生態總體更為完善一些，且在機器學習的集成和應用性暫時領先。但作為下一代大數據引擎的有力競爭者-Flink在流式計算上有明顯優勢，Flink在流式計算里屬於真正意義上的單條處理，每一條數據都觸發計算，而不是像Spark一樣的Mini Batch作為流式處理的妥協。Flink的容錯機制較為輕量，對吞吐量影響較小，而且擁有圖和調度上的一些優化，使得Flink可以達到很高的吞吐量。而Strom的容錯機制需要對每條數據進行ack，因此其吞吐量瓶頸也是備受詬病。

這裡引用一張圖來對常用的實時計算框架做個對比。

Flink特點

Flink是一個開源的分散式實時計算框架。Flink是有狀態的和容錯的，可以在維護一次應用程式狀態的同時無縫地從故障中恢復；它支援大規模計算能力，能夠在數千個節點上並發運行；它具有很好的吞吐量和延遲特性。同時，Flink提供了多種靈活的窗口函數。

1）狀態管理機制

Flink檢查點機制能保持exactly-once語義的計算。狀態保持意味著應用能夠保存已經處理的數據集結果和狀態。

2）事件機制

Flink支援流處理和窗口事件時間語義。事件時間可以很容易地通過事件到達的順序和事件可能的到達延遲流中計算出準確的結果。

3）窗口機制

Flink支援基於時間、數目以及會話的非常靈活的窗口機制（window）。可以訂製window的觸發條件來支援更加複雜的流模式。

4）容錯機制

Flink高效的容錯機制允許系統在高吞吐量的情況下支援exactly-once語義的計算。Flink可以準確、快速地做到從故障中以零數據丟失的效果進行恢復。

5）高吞吐、低延遲

Flink具有高吞吐量和低延遲（能快速處理大量數據）特性。下圖展示了Apache Flink和Apache Storm完成分散式項目計數任務的性能對比。

二、架構演變

初期架構

初期架構僅為計算與存儲兩層，新來的計算需求接入後需要新開發一個實時計算任務進行上線。重複模組的程式碼復用率低，重複率高，計算任務間的區別主要是集中在任務的計算指標口徑上。

在存儲層，各個需求方所需求的存儲路徑都不相同，計算指標可能在不通的存儲引擎上有重複，有計算資源以及存儲資源上的浪費情況。並且對於指標的計算口徑也是僅局限於單個任務需求里的，不通需求任務對於相同的指標的計算口徑沒有進行統一的限制於保障。各個業務方也是在不同的存儲引擎上開發數據獲取服務，對於那些專註於數據應用本身的團隊來說，無疑當前模式存在一些弊端。

後期架構

隨著數據體量的增加以及業務線的擴展，前期架構模式的弊端逐步開始顯現。從當初單需求單任務的模式逐步轉變為通用的數據架構模式。為此，我們開發了一些基於Flink框架的通用組件來支援數據的快速接入，並保證程式碼模式的統一性和維護性。在數據層，我們基於Clickhouse來作為我們數據倉庫的計算和存儲引擎，利用其支援多維OLAP計算的特性，來處理在多維多指標大數據量下的快速查詢需求。在數據分層上，我們參考與借鑒離線數倉的經驗與方法，構建多層實時數倉服務，並開發多種微服務來為數倉的數據聚合，指標提取，數據出口，數據品質，報警監控等提供支援。

整體架構分為五層：

1）接入層：接入原始數據進行處理，如Kafka、RabbitMQ、File等。

2）計算層：選用Flink作為實時計算框架，對實時數據進行清洗，關聯等操作。

3）存儲層： 對清洗完成的數據進行數據存儲，我們對此進行了實時數倉的模型分層與構建，將不同應用場景的數據分別存儲在如Clickhouse,Hbase,Redis,Mysql等存儲。服務中，並抽象公共數據層與維度層數據，分層處理壓縮數據並統一數據口徑。

4）服務層：對外提供統一的數據查詢服務，支援從底層明細數據到聚合層數據5min/10min/1hour的多維計算服務。同時最上層特徵指標類數據，如計算層輸入到Redis、Mysql等也從此數據介面進行獲取。

5）應用層：以統一查詢服務為支撐對各個業務線數據場景進行支撐。

監控報警：對Flink任務的存活狀態進行監控，對異常的任務進行郵件報警並根據設定的參數對任務進行自動拉起與恢復。根據如Kafka消費的offset指標對消費處理延遲的實時任務進行報警提醒。

數據品質：監控實時數據指標，對歷史的實時數據與離線hive計算的數據定時做對比，提供實時數據的數據品質指標，對超過閾值的指標數據進行報警。

三、數據處理流程

1、整體流程

整體數據從原始數據接入後經過ETL處理, 進入實時數倉底層數據表，經過配置化聚合微服務組件向上進行分層數據的聚合。根據不同業務的指標需求也可通過特徵抽取微服務直接配置化從數倉中抽取到如Redis、ES、Mysql中進行獲取。大部分的數據需求可通過統一數據服務介面進行獲取。

2、問題與挑戰

原始日誌數據因為各業務日誌的不同，所擁有的維度或指標數據並不完整。所以需要進行實時的日誌的關聯才能獲取不同維度條件下的指標數據查詢結果。並且關聯日誌的回傳周期不同，有在10min之內完成95%以上回傳的業務日誌，也有類似於激活日誌等依賴第三方回傳的有任務日誌，延遲窗口可能大於1天。並且最大日誌關聯任務的日均數據量在10億級別以上，如何快速處理與構建實時關聯任務的問題首先擺在我們面前。對此我們基於Flink框架開發了配置化關聯組件。對於不同關聯日誌的指標抽取，我們也開發了配置化指標抽取組件用於快速提取複雜的日誌格式。以上兩個自研組件會在後面的內容里再做詳細介紹。

1）回傳周期超過關聯窗口的日誌如何處理?

對於回傳晚的日誌，我們在關聯窗口內未取得關聯結果。我們採用實時+離線的方式進行數據回刷補全。實時處理的日誌我們會將未關聯的原始日誌輸出到另外一個暫存地(Kafka)，同時不斷消費處理這個未關聯的日誌集合，設定超時重關聯次數與超時重關聯時間，超過所設定任意閾值後，便再進行重關聯。離線部分，我們採用Hive計算昨日全天日誌與N天內的全量被關聯日誌表進行關聯，將最終的結果回寫進去，替換實時所計算的昨日關聯數據。

2）如何提高Flink任務性能？

① Operator Chain

為了更高效地分散式執行，Flink會儘可能地將operator的subtask鏈接（chain）在一起形成task。每個task在一個執行緒中執行。將operators鏈接成task是非常有效的優化：它能減少執行緒之間的切換，減少消息的序列化/反序列化，減少數據在緩衝區的交換，減少了延遲的同時提高整體的吞吐量。

Flink會在生成JobGraph階段，將程式碼中可以優化的運算元優化成一個運算元鏈（Operator Chains）以放到一個task（一個執行緒）中執行，以減少執行緒之間的切換和緩衝的開銷，提高整體的吞吐量和延遲。下面以官網中的例子進行說明。

圖中，source、map、[keyBy|window|apply]、sink運算元的並行度分別是2、2、2、2、1，經過Flink優化後，source和map運算元組成一個運算元鏈，作為一個task運行在一個執行緒上，其簡圖如圖中condensed view所示，並行圖如parallelized view所示。運算元之間是否可以組成一個Operator Chains，看是否滿足以下條件：

• 上下游運算元的並行度一致；
• 下游節點的入度為1；
• 上下游節點都在同一個 slot group 中；
• 下游節點的 chain 策略為 ALWAYS；
• 上游節點的 chain 策略為 ALWAYS 或 HEAD；
• 兩個節點間數據分區方式是 forward；
• 用戶沒有禁用 chain。

② Flink非同步IO

流式計算中，常常需要與外部系統進行交互。而往往一次連接中你那個獲取連接等待通訊的耗時會佔比較高。下圖是兩種方式對比示例：

圖中棕色的長條表示等待時間，可以發現網路等待時間極大地阻礙了吞吐和延遲。為了解決同步訪問的問題，非同步模式可以並發地處理多個請求和回復。也就是說，你可以連續地向資料庫發送用戶a、b、c等的請求，與此同時，哪個請求的回復先返回了就處理哪個回復，從而連續的請求之間不需要阻塞等待，如上圖右邊所示。這也正是 Async I/O 的實現原理。

③ Checkpoint優化

Flink實現了一套強大的checkpoint機制，使它在獲取高吞吐量性能的同時，也能保證Exactly Once級別的快速恢復。

首先提升各節點checkpoint的性能考慮的就是存儲引擎的執行效率。Flink官方支援的三種checkpoint state存儲方案中，Memory僅用於調試級別，無法做故障後的數據恢復。其次還有Hdfs與Rocksdb，當所做Checkpoint的數據大小較大時，可以考慮採用Rocksdb來作為checkpoint的存儲以提升效率。

其次的思路是資源設置，我們都知道checkpoint機制是在每個task上都會進行，那麼當總的狀態數據大小不變的情況下，如何分配減少單個task所分的的checkpoint數據變成了提升checkpoint執行效率的關鍵。

最後，增量快照. 非增量快照下，每次checkpoint都包含了作業所有狀態數據。而大部分場景下，前後checkpoint里，數據發生變更的部分相對很少，所以設置增量checkpoint，僅會對上次checkpoint和本次checkpoint之間狀態的差異進行存儲計算，減少了checkpoint的耗時。

3）如何保障任務的穩定性?

在任務執行過程中，會遇到各種各樣的問題，導致任務異常甚至失敗。所以如何做好異常情況下的恢復工作顯得異常重要。

① 設定重啟策略

Flink支援不同的重啟策略，以在故障發生時控制作業如何重啟。集群在啟動時會伴隨一個默認的重啟策略，在沒有定義具體重啟策略時會使用該默認策略。如果在工作提交時指定了一個重啟策略，該策略會覆蓋集群的默認策略。

默認的重啟策略可以通過 Flink 的配置文件 flink-conf.yaml 指定。配置參數 restart-strategy 定義了哪個策略被使用。

常用的重啟策略：

• 固定間隔(Fixed delay)；
• 失敗率(Failure rate)；
• 無重啟(No restart)。

② 設置HA

Flink在任務啟動時指定HA配置主要是為了利用Zookeeper在所有運行的JobManager實例之間進行分散式協調.Zookeeper通過leader選取和輕量級一致性的狀態存儲來提供高可用的分散式協調服務。

③ 任務監控報警平台

在實際環境中，我們遇見過因為集群狀態不穩定而導致的任務失敗。在Flink1.6版本中，甚至遇見過任務出現假死的情況，也就是Yarn上的job資源依然存在，而Flink任務實際已經死亡。為了監測與恢復這些異常的任務，並且對實時任務做統一的提交、報警監控、任務恢復等管理，我們開發了任務提交與管理平台。通過Shell拉取Yarn上Running狀態與Flink Job狀態的列表進行對比，心跳監測平台上的所有任務，並進行告警、自動恢復等操作。

④ 作業指標監控

Flink任務在運行過程中，各Operator都會產生各自的指標數據，例如，Source會產出numRecordIn、numRecordsOut等各項指標資訊，我們會將這些指標資訊進行收集，並展示在我們的可視化平台上。指標平台如下圖：

⑤ 任務運行節點監控

我們的Flink任務都是運行在Yarn上，針對每一個運行的作業，我們需要監控其運行環境。會收集JobManager及TaskManager的各項指標。收集的指標有jobmanager-fullgc-count、jobmanager-younggc-count、jobmanager-fullgc-time、jobmanager-younggc-time、taskmanager-fullgc-count、taskmanager-younggc-count、taskmanager-fullgc-time、taskmanager-younggc-time等，用於判斷任務運行環境的健康度，及用於排查可能出現的問題。監控介面如下：

四、數據關聯組件

1、如何選擇關聯方式？

1）Flink Table

從Flink的官方文檔，我們知道Flink的編程模型分為四層，sql是最高層的api, Table api是中間層，DataSteam/DataSet Api是核心，stateful Streaming process層是底層實現。

剛開始我們直接使用Flink Table做為數據關聯的方式，直接將接入進來的DataStream註冊為Dynamic Table後進行兩表關聯查詢，如下圖：

但嘗試後發現在做那些日誌數據量大的關聯查詢時往往只能在較小的時間窗口內做查詢，否則會超過datanode節點單台記憶體限制，產生異常。但為了滿足不同業務日誌延遲到達的情況，這種實現方式並不通用。

2）Rocksdb

之後，我們直接在DataStream上進行處理，在CountWindow窗口內進行關聯操作，將被關聯的數據Hash打散後存儲在各個datanode節點的Rocksdb中，利用Flink State原生支援Rocksdb做Checkpoint這一特性進行運算元內數據的備份與恢復。這種方式是可行的，但受制於Rocksdb集群物理磁碟為非SSD的因素，這種方式在我們的實際線上場景中關聯耗時較高。

3）外部存儲關聯

如Redis類的KV存儲的確在查詢速度上提升不少，但類似廣告日誌數據這樣單條日誌大小較大的情況，會佔用不少寶貴的機器記憶體資源。經過調研後，我們選取了Hbase作為我們日誌關聯組件的關聯數據存儲方案。

為了快速構建關聯任務，我們開發了基於Flink的配置化組件平台，提交配置文件即可生成數據關聯任務並自動提交到集群。下圖是任務執行的處理流程。

示意圖如下：

下圖是關聯組件內的執行流程圖：

2、問題與優化

1）加入Interval Join

隨著日誌量的增加，某些需要進行關聯的日誌數量可能達到日均十幾億甚至幾十億的量級。前期關聯組件的配置化生成任務的方式的確解決了大部分線上業務需求，但隨著進一步的關聯需求增加，Hbase面臨著巨大的查詢壓力。在我們對Hbase表包括rowkey等一系列完成優化之後，我們開始了對關聯組件的迭代與優化。

第一步，減少Hbase的查詢。我們使用Flink Interval Join的方式，先將大部分關聯需求在程式內部完成，只有少部分仍需查詢的日誌會去查詢外部存儲(Hbase). 經驗證，以請求日誌與實驗日誌關聯為例，對於設置Interval Join窗口在10s左右即可減少80%的hbase查詢請求

① Interval Join的語義示意圖

• 數據JOIN的區間 – 比如時間為3的EXP會在IMP時間為[2, 4]區間進行JOIN；
• WaterMark – 比如圖示EXP一條數據時間是3，IMP一條數據時間是5，那麼WaterMark是根據實際最小值減去UpperBound生成，即：Min(3,5)-1 = 2；
• 過期數據 – 出於性能和存儲的考慮，要將過期數據清除，如圖當WaterMark是2的時候時間為2以前的數據過期了，可以被清除。

② Interval Join內部實現邏輯

③ Interval Join改造

因Flink原生的Intervak Join實現的是Inner Join,而我們業務中所需要的是Left Join，具體改造如下：

• 取消右側數據流的join標誌位；
• 左側數據流有join數據時不存state。

2）關聯率動態監控

在任務執行中，往往會出現意想不到的情況，比如被關聯的數據日誌出現缺失，或者日誌格式錯誤引發的異常，造成關聯任務的關聯率下降嚴重。那麼此時關聯任務雖然繼續在運行，但對於整體數據品質的意義不大，甚至是反向作用。在任務進行恢復的時，還需要清除異常區間內的數據，將Kafka Offset設置到異常前的位置再進行處理。

故我們在關聯組件的優化中，加入了動態監控，下面示意圖：

• 關聯任務中定時探測指定時間範圍 Hbase是否有最新數據寫入，如果沒有，說明寫Hbase任務出現問題，則終止關聯任務；
• 當寫Hbase任務出現堆積時，相應的會導致關聯率下降，當關聯率低於指定閾值時終止關聯任務；
• 當關聯任務終止時會發出告警，修復上游任務後可重新恢復關聯任務，保證關聯數據不丟失。

五、數據清洗組件

為了快速進行日誌數據的指標抽取，我們開發了基於Flink計算平台的指標抽取組件Logwash。封裝了基於Freemaker的模板引擎做為日誌格式的解析模組，對日誌進行提取，算術運算，條件判斷，替換，循環遍歷等操作。

下圖是Logwash組件的處理流程：

組件支援文本與Json兩種類型日誌進行解析提取，目前該清洗組件已支援微博廣告近百個實時清洗需求，提供給運維組等第三方非實時計算方向人員快速進行提取日誌的能力。

配置文件部分示例：

六、FlinkStream組件庫

Flink中DataStream的開發，對於通用的邏輯及相同的程式碼進行了抽取，生成了我們的通用組件庫FlinkStream。FlinkStream包括了對Topology的抽象及默認實現、對Stream的抽象及默認實現、對Source的抽象和某些實現、對Operator的抽象及某些實現、Sink的抽象及某些實現。任務提交統一使用可執行Jar和配置文件，Jar會讀取配置文件構建對應的拓撲圖。

1、Source抽象

對於Source進行抽象，創建抽象類及對應介面，對於Flink Connector中已有的實現，例如kafka,Elasticsearch等，直接創建新class並繼承介面，實現對應的方法即可。對於需要自己去實現的connector，直接繼承抽象類及對應介面，實現方法即可。目前只實現了KafkaSource。

2、Operator抽象

與Source抽象類似，我們實現了基於Stream到Stream級別的Operator抽象。創建抽象Operate類，抽象Transform方法。對於要實現的Transform操作，直接繼承抽象類，實現其抽象方法即可。目前實現的Operator，直接按照文檔使用。如下：

3、Sink抽象

針對Sink，我們同樣創建了抽象類及介面。對Flink Connector中已有的Sink進行封裝。目前可通過配置進行數據輸出的Sink。目前以實現和封裝的Sink組件有：Kafka、Stdout、Elasticsearch、Clickhouse、Hbase、Redis、MySQL。

4、Stream抽象

創建Stream抽象類及抽象方法buildStream，用於構建StreamGraph。我們實現了默認的Stream，buildStream方法讀取Source配置生成DataStream，通過Operator配置列表按順序生成拓撲圖，通過Sink配置生成數據寫出組件。

5、Topology抽象

對於單Stream，要處理的邏輯可能比較簡單，主要讀取一個Source進行數據的各種操作並輸出。對於複雜的多Stream業務需求，比如多流Join，多流Union、Split流等，因此我們多流業務進行了抽象，產生了Topology。在Topology這一層可以對多流進行配置化操作。對於通用的操作，我們實現了默認Topology，直接通過配置文件就可以實現業務需求。對於比較複雜的業務場景，用戶可以自己實現Topology。

6、配置化

我們對抽象的組件都是可配置化的，直接通過編寫配置文件，構造任務的運行拓撲結構，啟動任務時指定配置文件。

• 正文文本框Flink Environment配置化，包括時間處理類型、重啟策略，checkpoint等；
• Topology配置化，可配置不同Stream之間的處理邏輯與Sink；
• Stream配置化，可配置Source，Operator列表，Sink。

配置示例如下：

run_env:      timeCharacteristic: "ProcessingTime" #ProcessingTimeIngestionTimeEventTime    restart: # 重啟策略配置      type: # noRestart, fixedDelayRestart, fallBackRestart, failureRateRestart    checkpoint: # 開啟checkpoint      type: "rocksdb" #  streams:    impStream:  #粉絲經濟曝光日誌      type: "DefaultStream"      config:        source:          type: "Kafka011" # 源是kafka011版本          config:          parallelism: 20        operates:          -            type: "StringToMap"            config:          -            type: "SplitElement"            config:          ...          -            type: "SelectElement"            config:  transforms:    -      type: "KeyBy"      config:    -      type: "CountWindowWithTimeOut"  #Window需要和KeyBy組合使用      config:    -      type: "SplitStream"      config:    -      type: "SelectStream"      config:  sink:    -      type: Kafka      config:    -      type: Kafka      config:

7、部署

在實時任務管理平台，新建任務，填寫任務名稱，選擇任務類型（Flink）及版本，上傳可執行Jar文件，導入配置或者手動編寫配置，填寫JobManager及TaskManager記憶體配置，填寫並行度配置，選擇是否重試，選擇是否從checkpoint恢復等選項，保存後即可在任務列表中啟動任務，並觀察啟動日誌用於排查啟動錯誤。

七、FlinkSQL擴展

SQL語言是一門聲明式的，簡單的，靈活的語言，Flink本身提供了對SQL的支援。Flink1.6版本和1.8版本對SQL語言的支援有限，不支援建表語句，不支援對外部數據的關聯操作。因此我們通過Apache Calcite對Flink SQL API進行了擴展，用戶只需要關心業務需求怎麼用SQL語言來表達即可。

1、支援創建源表

擴展了支援創建源表SQL，通過解析SQL語句，獲取數據源配置資訊，創建對應的TableSource實例，並將其註冊到Flink environment。示例如下：

2、支援創建維表

使用Apache Calcite對SQL進行解析，通過維表關鍵字識別維表，使用RichAsyncFunction運算元非同步讀取維表數據，並通過flatMap操作生成關聯後的DataStream，然後轉換為Table註冊到Flink Environment。示例如下：

3、支援創建視圖

使用sqlQuery方法，支援從上一層表或者視圖中創建視圖表，並將新的視圖表註冊到Flink Environment。創建語句需要按照順序寫，比如myView2是從視圖myView1中創建的，則myView1創建語句要在myView2語句前面。如下：

4、支援創建結果表

支援創建結果表，通過解析SQL語句，獲取配置資訊，創建對應的AppendStreamTableSink或者UpsertStreamTableSink實例，並將其註冊到Flink Environment。示例如下：

5、支援自定義UDF

支援自定義UDF函數，繼承ScalarFunction或者TableFunction。在resources目錄下有相應的UDF資源配置文件，默認會註冊全部可執行Jar包中配置的UDF。直接按照使用方法使用即可。

6、部署

部署方式同FlinkStream組件。

八、實時數據倉庫的構建

為了保證實時數據的統一對外出口以及保證數據指標的統一口徑，我們根據業界離線數倉的經驗來設計與構架微博廣告實時數倉。

1、分層概覽

數據倉庫分為三層，自下而上為：數據引入層（ODS，Operation Data Store）、數據公共層（CDM，Common Data Model）和數據應用層（ADS，Application Data Service）

數據引入層（ODS，Operation Data Store）：將原始數據幾乎無處理的存放在數據倉庫系統，結構上與源系統基本保持一致，是數據倉庫的數據准。

數據公共層（CDM，Common Data Model，又稱通用數據模型層）：包含DIM維度表、DWD和DWS，由ODS層數據加工而成。主要完成數據加工與整合，建立一致性的維度，構建可復用的面向分析和統計的明細事實表，以及匯總公共粒度的指標。

公共維度層（DIM）：基於維度建模理念思想，建立整個企業的一致性維度。降低數據計算口徑和演算法不統一風險。

公共維度層的表通常也被稱為邏輯維度表，維度和維度邏輯表通常一一對應。

公共匯總粒度事實層（DWS，Data Warehouse Service）：以分析的主題對象作為建模驅動，基於上層的應用和產品的指標需求，構建公共粒度的匯總指標事實表，以寬表化手段物理化模型。構建命名規範、口徑一致的統計指標，為上層提供公共指標，建立匯總寬表、明細事實表。

公共匯總粒度事實層的表通常也被稱為匯總邏輯表，用於存放派生指標數據。

明細粒度事實層（DWD，Data Warehouse Detail）：以業務過程作為建模驅動，基於每個具體的業務過程特點，構建最細粒度的明細層事實表。可以結合企業的數據使用特點，將明細事實表的某些重要維度屬性欄位做適當冗餘，也即寬表化處理。

明細粒度事實層的表通常也被稱為邏輯事實表。

數據應用層（ADS，Application Data Service）：存放數據產品個性化的統計指標數據。根據CDM與ODS層加工生成。

2、詳細分層模型

對於原始日誌數據，ODS層幾乎是每條日誌抽取欄位後進行保留，這樣便能對問題的回溯與追蹤。在CDM層對ODS的數據僅做時間粒度上的數據壓縮，也就是在指定時間切分窗口裡，對所有維度下的指標做聚合操作，而不涉及業務性的操作。在ADS層，我們會有配置化抽取微服務，對底層數據做訂製化計算和提取，輸出到用戶指定的存儲服務里。