摘要：本文由好未來資深資料平台工程師毛祥溢分享，主要介紹批流融合在教育行業的實踐。内容包括兩部分，第一部分是好未來在做實時平台中的幾點思考，第二部分主要分享教育行業中特有資料分析場景。大綱如下：
背景介紹
好未來 T-Streaming 實時平台
K12 教育典型分析場景
展望與規劃
1.背景介紹
好未來介紹
好未來是一家 2003 年成立教育科技公司，旗下有品牌學而思，現在大家聽說的學而思培優、學而思網校都是該品牌的衍生，2010 年公司在美國納斯達克上市，2013 年更名為好未來。2016 年，公司的業務範圍已經覆寫負一歲到 24 歲的使用者。目前公司主營業務單元有智慧教育、教育領域的開放平台、K12 教育以及海外留學等業務。
好未來資料中台全景圖
上圖為好未來資料中台的全景圖，主要分為三層：
● 第一層是資料賦能層
● 第二層是全域資料層
● 第三層是資料開發層
首先，資料賦能層。主要是商業智能、智慧決策的應用，包括一些資料工具、資料能力以及專題分析體系，資料工具主要包括埋點資料分析工具、AB 測試工具、大屏工具；資料能力分析主要包括未來畫像服務、未來增長服務、未來使用者服務以及新校區的選址服務；專題分析體系主要包企業經營類專題分析等等。
其次，資料全域層。我們期望将全集團所有的事業部的資料進行深入的拉通和融合，打通不同業務線、産品線的使用者池，進而盤活全集團的資料。具體的手段是 IDMapping，将裝置 id、自然人、家庭三個層級的 id 映射關系挖掘出來，将不同産品上的使用者資料關聯起來。這樣就能夠形成一個打的使用者池，友善我們更好的賦能使用者。
最後，資料開發層。資料開發通過一些列的平台承載了全集團所有的資料開發工程，主要包括資料內建、資料開發、資料品質、資料服務、資料治理等服務。我們今天要分享的實時平台就是在資料開發中。
2.好未來 T-Streaming 實時平台
實時平台建構前的訴求
實時平台在建構之初，我們梳理了四個重要的訴求。
● 第一個訴求是期望有一套統一的叢集，通過提供多租戶，資源隔離的方式提高資源使用率，解決多個事業部多套叢集的問題。
● 第二個訴求是期望通過平台的方式降低實時資料開發的門檻，進而能夠覆寫更多的開發者。
● 第三個訴求是期望能夠提供通用場景的解決解方案，提高項目的複用性，避免每個事業部都開發相同場景的分析工具。
● 第四個訴求是對作業進行全方位的生命周期管理，包括中繼資料和血緣，一旦有一個作業出現異常，我們可以快速分析和定位影響範圍。
實時平台功能概述
現在我們平台已經是一個一站式的實時資料分析平台，包括了資料內建、資料開發、作業保障、資源管理、資料安全等功能。
● 在資料內建方面，我們支援資料庫、埋點資料、服務端日志資料的內建，為了能夠提高資料內建的效率，我們提供了很多的通用模闆作業，使用者隻需要配置即可快速實作資料的內建。
● 在資料開發方面，我們支援兩種方式的作業開發，一種是 Flink SQL 作業開發、一種是 Flink Jar 包托管，在 Flink SQL 開發上我們内置了很多 UDF 函數，比如可以通過 UDF 函數實作維表 join，也支援使用者自定義 UDF，并且實作了 UDF 的熱加載。除此之外，我們也會記錄使用者在作業開發過程中的中繼資料資訊，友善血緣系統的建設。
● 在作業保障方面，我們支援作業狀态監控、異常告警、作業失敗之後的自動拉起，作業自動拉起我們會自動選擇可用的 checkpoint 版本進行拉起，同時也支援作業在多叢集之間的切換。
● 在資源管理方面，我們支援平台多租戶，每個租戶使用 namespace 進行隔離、實作了不同僚業部、不同使用者、不同版本的 Flink 用戶端隔離、實作了計算資源的隔離。
● 在資料安全方面，我們支援角色權限管理、表級别權限管理、操作審計日志查詢等功能。
以上就是我們平台的功能，在賦能業務的同時，我們也還在快速疊代中，期望平台簡單好用，穩定可信賴。
實時平台的批流融合
接下來說一下平台建設中的一些實踐，第一個是批流融合。
我們先理清楚批流融合是什麼？
批流融合可以分為兩個概念，一個是 Flink 提出的批流融合，具體的了解就是一個 Flink SQL 既可以作用于流資料、也可以作用于批資料，通過保證計算引擎一緻進而減少結果資料的差異，這是一個技術層面上的批流融合。另個一概念是我們内部提出來的，那就是架構層面的批流融合。具體的操作手法就是通過 Flink 作業保證資料倉庫 ODS 層的實時化，然後提供小時級别、分鐘級别的排程，進而提高資料分析的實時化。
為什麼我們會提出架構上的批流融合，主要我們看到行業發展的兩個趨勢。
● 第一個趨勢是資料內建的實時化群組件化，比如 Flink 內建 Hive、Flink CDC 的持續完善和增強，這樣我們做資料內建的時候就會變得非常簡單。
● 第二個趨勢是實時 OLAP 引擎越來越成熟，比如 Kudu+impala、阿裡雲的 Hologres、湖倉一體的方案。
這兩個趨勢讓使用者開發實時資料會變得越來越簡單，使用者隻需要關注 SQL 本身就可以。
如上圖所示，我們有三個類型的實時數倉，一個是基于 Hive 的、一個是基于實時 OLAP 引擎的、一個是基于 Kafka 的。其中，藍色線條就是我們 ODS 層實時化的具體實作。我們提供了一個統一的工具，可以将實時的将資料寫入到 Hive、實時 OLAP 引擎、當然還有 Kafka。這個工具使用起來比較簡單，如果是 MySQL 資料的同步，使用者隻需要輸入資料庫名稱和表名就可以了。
通過 ODS 層實時化的工具，我們就可以在 Hive、實時 OLAP 引擎、Kafka 中建構實時數倉。
● 如果是 Hive 實時數倉，我們會使用 Flink 将實時的增量資料寫入到 ODS 層，然後提供一個定時 merge 的腳本，用來 merge 增量資料和曆史資料，進而保證 ODS 層的資料是最新最全的。配合 airflow 小時級别的排程能力，使用者就可以得到一個小時級别的數倉了。
● 如果是類似于 Kudu / Hologres 這樣的實時 OLAP 引擎，我們會先把離線資料從 Hive 中導入到實時 OLAP 引擎中，然後使用 Flink 将實時的增量資料寫入到 ODS 層，寫入的方式推薦使用 upsert 這樣的特性，這樣使用者就能夠得到一個純實時的數倉了。配合 airflow 分鐘級别的排程能力，使用者就可以得到一個分鐘級别的數倉了。
● 基于 Kafka 建構實時數倉，就是非常經典的架構了，開發成本也比較高一些，除了必須要秒級更新的分析場景，我們不太建議使用者使用。當然在 2021 年的時候，我們也會去做 Flink 批流一體解決方案，讓使用者有更多選擇方式的同時，讓整個實時數倉變得更加簡單。
以上就是我們對批流融合的思考和實踐，通過這種架構層面的批流融合，原來需要開發一個月的實時需求，現在 2 天就差不多能完成。大大降低了開發實時資料的門檻，提高了資料分析的效率。
實時平台 ODS 層實時化
說一下 ODS 層實時化我們具體是怎麼做的。
要想把 ODS 層資料實時化，我們需要解決兩個問題，第一個是離線資料的初始化問題，第二個是增量資料如何寫入的問題。離線資料導入比較好做，如果資料源是 MySQL，我們可以使用 DataX 或者 Spark 作業的方式将 MySQL 的全量資料導入到 Hive 中，而實時增量資料的寫入我們需要有兩個步驟，第一個步驟是将 MySQL 的 binlog 采集到 Kafka，第二個步驟是将 Kafka 的資料使用Flink作業導入到 Hive。這樣算下來，要解決 ODS 層實時化的問題，我們就需要一個離線初始化的作業，一個增量資料采集的作業，一個增量資料寫入的作業，也就是需要 3 個作業。
在我們的平台上，我們對 ODS 層的 3 個作業進行了封裝和統一排程，使用者隻需要輸入一個資料庫名稱和表的名稱就能完成 ODS 層實時化的工作。
以上就是我們批流融合中 ODS 層實時化的實作過程。
實時平台 Flink SQL 開發流程
image.png我們另外一個實踐，就是對 Flink SQL 的作業封裝。先看一下，在我們平台上進行 Flink SQL 開發的整體流程。
從左往右看，資料源中的資料會通過 Maxwell、canal 這樣的工具采集到 Kafka，采集到 Kafka 的原始資料格式并不是統一的，是以我們需要将 Kafka 中的資料進行統一格式化處理，我們預設支援埋點資料格式、canal 資料格式、maxwell 資料的解析，也支援使用者自己上傳 Jar 包進行資料解析，解析得到的标準化資料就會再次發送到 Kafka。
然後我們會使用 Flink SQL 作業來消費 Kafka 的資料，進行 SQL 腳本的開發。這裡的 SQL 腳本開發和原生的 Flink SQL 的腳本開發有一點不一樣，原生的 SQL 腳本開發使用者需要編寫 Source 資訊、Sink 資訊，在我們平台上使用者隻需要寫具體的 SQL 邏輯就可以了。
那使用者寫完 SQL 之後，會将 SQL 作業資訊送出到我們封裝好的 Flink SQL 執行作業上，最後通過我們封裝的 SQL 引擎将作業送出的 Flink 叢集上去運作。後面将介紹我們是怎麼封裝的。
以上就是在我們平台上進行 Flink SQL 開發的流程，出了 Flink 作業本身的開發和送出，平台也會保留與作業有關的各種輸入、輸出的 schema 資訊。比如業務資料庫表的 schema 資訊，經過同意加工之後的 schema 資訊，資料輸出的表的 schema 資訊，通過這些記錄，後期我們排查問題的時候就能夠快速梳理出作業的來龍去脈和影響範圍。
實時平台 Flink SQL 開發過程
在我們平台上開發 Flink SQL 作業，隻需要三個步驟：
● 第一個步驟确認 Kafka 的 Topic 是否已經注冊過了，如果沒有注冊就需要使用者手動注冊下，完成注冊後，我們會把 Topic 的資料解析出來，将字段資訊儲存起來。
● 第二步使使用者編寫 SQL，剛才說過，使用者隻需要寫具體的 SQL 邏輯，不需要寫 Source 和 Sink 資訊。
● 第三步是使用者指定将資料輸出到哪裡，現在平台可以支援同時指定多個 Sink 儲存設備，比如将計算好的資料同時寫入到 Hive、Holo 等存儲。
通過以上三個步驟的配置，使用者就可以送出作業了。
接下來說一下，我們是怎麼做的，我把整個執行過程分為 2 個階段 10 個步驟。
第一個階段就是作業準備階段，第二個階段就是 SQL 執行階段。
■ 作業準備階段
● 第一步，使用者在頁面資料 SQL 和指定 Sink 資訊。
● 第二步，SQL 解析及校驗過程，當使用者送出 SQL 時，我們會對 SQL 進行解析，看看 SQL 中用到的 Source 表和 UDF 是否在平台中注冊過。
● 第三步，推測建表，我們會先運用下使用者的 SQL，然後得到 SQL 的傳回結果，根據結果資料生成一些建表語句，最後通過程式自動到目标 Sink 存儲上去建表。
● 第四步，拼裝 Flink SQL 的腳本檔案，得到一個有 Source、SQL、Sink 三要素的腳本檔案。
● 第五步，作業送出，這裡會把 Flink SQL 檔案送出到我們自己執行引擎中。
■ SQL 執行階段
● 第一步是會初始化 StreamTableAPI，然後使用 connect 方法注冊 Kafka Source，Kafka 的 Source 資訊需要指定資料解析的規則和字段的 schema 資訊，我們會根據中繼資料自動生成。
● 第二步是使用 StreamTableAPI 注冊 SQL 中使用到的維表和 UDF 函數，UDF 函數包括使用者自己上傳的 UDF 函數。
● 第三步是使用 StreamTable API 執行 SQL 語句，如果有視圖也可以執行視圖。
● 第四步是一個比較關鍵的步驟，我們會把 StreamTabAPI 轉成 DataStream API。
● 第五步就是在 DataStream 的基礎上 addSink 資訊了。
以上是兩個階段的執行過程，通過第二個階段，使用者的 SQL 作業就會真正的運作起來。
實時平台原生作業與模闆任務
上面分享了我們的 Flink SQL 作業如何開發和運作，接下來說一下我們平台對 JAR 包類型作業的支援。
在我們平台上，我們支援使用者自己上傳 JAR 包作業，然後在我們平台上進行管理。與此同時，為了提高代碼通常場景的複用性，我們開發了很多模闆作業，比如支援 Maxwell 采集的 binlog 直接寫入到 Hive、Kudu、Holo 等儲存設備，支援阿裡雲 SLS 日志寫入到各種 OLAP 引擎。
實時平台混合雲部署方案
講一下混合雲部署方案和平台技術架構。
我們平台現在支援将作業送出到阿裡雲機房、自建機房中，并且作業可以在兩個機房中來回切換。為了要有這個功能呢?
今年年初，随着疫情的爆發，網際網路線上教育湧入了大量的流量，為了應對暴增的流量，春節期間我們采購了上千台機器進行緊急的部署和上線，後來疫情穩定住了之後，這些機器的使用率就比較低了，為了解決這個問題，我們平台就支援了混合雲部署方案，高峰期的時候作業可以遷移到阿裡雲上運作，平常就在自己的叢集上運作，既節約了資源又保證了彈性擴容。
實時平台技術架構
接下來說一下平台的技術架構。
我們是一個前後端分離的項目，前端使用 vue+elmentui、服務端使用 springboot，不同的機房裡面我們會部署一個後端服務的執行個體。任務送出到不同的機房主要通過轉發層的 nginx+lua 來實作的。平台上任務的送出、暫停、下線操作，都是通過驅動層來完成的，驅動層主要是一些 shell 腳本。最後就是用戶端了，在用戶端上我們做了 Namespace/使用者/Flink 版本的隔離。
3.K12 教育典型分析場景
續報業務介紹
我們聊一個具體的案例，案例是 K12 教育行業中典型的分析場景，使用者續報業務。
先說下什麼是續報，續報就是重複購買，使用者購買了一年的課程，我們期望使用者購買二年的課程。為了使用者購買課程，我們會有一個集中的時間段用來做續報，每次持續一周左右，一年四次。
因為續報周期比較集中，時間比較短暫，每次做續報業務老師對實時續報資料的需求就特别迫切。
為此我們做了一個通用的續報解決方案，來支援各事業部的續報動作。要做實時續報，有幾個挑戰。
● 第一個挑戰是計算一個使用者的訂單是否是續報，需要依賴這個使用者曆史上所有的訂單，也就是需要曆史資料參與計算。
● 第二個挑戰就是一個訂單的變化會影響其它訂單的變化，是一個連鎖效應。比如使用者有 5 個訂單，編号為 345 的訂單都是續報狀态，如果使用者取消了編号為 3 的訂單，訂單 4 和訂單5的續報狀态就需要重新計算。
● 第三個挑戰是次元變化很頻繁，比如使用者上午的分校狀态是北京，下午的分校狀态可能就是上海，上午的輔導老師是張三，下午的輔導老師就是李四，頻繁變化的次元給實時彙總資料帶來了挑戰。
依賴曆史資料、訂單改變的連鎖效應、頻繁變化的次元，這些挑戰如果單個看都不算什麼，如果放在一起就會變得比較有意思了。
實時續報解決方案
先說下整體架構，我們采用的批流融合方式來做的，分成兩條線，一條線是分鐘級實時續報資料計算，一條是秒級實時續報資料計算。計算好的資料放在 MYSQL 中，用來做大屏和 BI 看闆。
先看下藍色的這條線，我們會把 Hive 中的離線資料導入到 Kudu 中，離線資料都是計算好的訂單寬表。然後會使用 Flink 作業把新增的訂單做成寬表寫入到 Kudu 中，這樣 Kudu 裡面就會有最新最全的資料。配合 4 分鐘的排程，我們就提供了分鐘級的實時續報資料。
在看第一條橙色的線條，這條線上有兩個 Flink 作業，一個是 ETL Job，一個是 Update Job。
ETL job 會負責靜态次元的拼接與續報狀态的計算，靜态次元拼接我們是直接通路 MySQL，然後緩存在 JVM 中。續報狀态的計算需要依賴曆史資料，ETL Job 會将所有的訂單資料加載到 JVM 中，具體的實作方法是我們自定義了一個 partitioncustom 方法，對所有的曆史資料進行了分片，下遊的每個 Task 緩存一個分片的資料。通過将資料加載到記憶體中，我們大大的加快了 Flink 實時計算的速度。
ETL Job 的計算的資料，會有兩個輸出，一個是輸出到 Kudu，用來保證 Kudu 中的資料最新最全，兩個一個資料是 Kafka，Kafka 中有一個 Topic 記錄的是是目前訂單的變化導緻了哪些訂單或者次元變化的資訊。
接在 Kafka 後面的程式就是 Update Job，專門用來處理受影響的訂單或者次元，直接去修改 MySQL 中相關的統計資料。
這樣我們就通過 2 個 Flink 作業實作的實時續報的計算。
最下面的一條線是實時次元的資料變更的處理，次元變更的資料會發送到 Kafka中，然後使用 Flink 進行處理，看看次元的變化影響了哪些資料的統計，最後将受影響的訂單發送到受影響的 Topic 中，由 Update Job 來重新計算。
以上就是我們實時續報的整體解決方案，如果有教育行業的朋友聽到這個分享，或許可以參考下。
實時續報穩定性保障
我們看看這個通用的解決方案上線之後有哪些保障。
● 第一個保障是異地雙活，我們在阿裡雲和自建機房都部署了一套續報程式，如果其中一套有異常，我們切換前端接口就可以了。如果兩個機房的程式都挂了，我們重零開始啟動程式，也隻需要 10 分鐘。
● 第二個保障是作業容錯，我們有兩個 Flink 作業，這兩個作業随停随啟，不影響資料的準确性。另外一點就是我們緩存了所有訂單資料在 JVM 中，如果資料量暴漲，我們隻需要改變 ETL 程式的并行度就可以，不用擔心 JVM 記憶體溢出。
● 第三個保障是作業監控，我們支援作業的異常告警和失敗後的自動拉起，也支援消費資料延遲告警。
通過以上保障措施，實時續報程式經過了幾次續報周期，都比較平穩，讓人很省心。
4.展望與規劃
上述内容詳細介紹了好未來目前業務及技術方案，總結而言我們通過多租戶實作各事業部資源隔離、通過批流融合的架構方案解決分析實時化、通過 ODS 層實時化解決資料源到 OLAP 的資料內建問題、通過 Flink SQL 封裝降低實時資料開發門檻、通過模闆任務提供通用場景解決方案、通過混合雲部署方案解決資源的彈性擴容、通過實時續報解決方案覆寫相同場景的資料分析。
最後，來看一下我們展望和規劃。接下來我們要繼續深化批流融合，強化混合雲部署，提高資料分析的時效性和穩定性。支援算法平台的實時化，資料應用的實時化，提高資料決策的時效性。