Flink 1.12 資源管理新特性回顧

本文由社群志願者陳政羽整理，Apache Flink Committer、阿裡巴巴技術專家宋辛童，Apache Flink Contributor、阿裡巴巴進階開發工程師郭旸澤分享，主要介紹 Flink 1.12 資源管理的一些特性。内容主要分為 4 部分：

記憶體管理

資源排程

擴充資源架構

未來規劃

GitHub 位址

https://github.com/apache/flink

歡迎大家給 Flink 點贊送 star~

一、記憶體管理

首先回顧 Flink 的記憶體模型變遷。下圖左邊分别為 Flink 1.10、Flink 1.11 引入的新的記憶體模型。盡管涉及的子產品較多，但 80% - 90% 的使用者僅需關注真正用于任務執行的 Task Heap Memory、Task Off-Heap Memory、Network Memory、Managed Memory 四部分。

其它子產品大部分是 Flink 的架構記憶體，正常不需要調整，即使遇到問題也可以通過社群文檔來解決。除此之外，“一個作業究竟需要多少記憶體才能滿足實際生産需求” 也是大家不得不面臨的問題，比如其他名額的功能使用、作業是否因為記憶體不足影響了性能，是否存在資源浪費等。

針對上述内容，社群在 Flink 1.12 版本提供了一個全新的，關于 Task manager 和 Job

manager 的 Web UI。

在新的 Web UI 中，可以直接将每一項監控名額配置值、實際使用情況對應到記憶體模型中進行直覺的展示。在此基礎上，可以更清楚的了解到作業的運作情況、該如何調整、用哪些配置參數調整等 (社群也有相應的文檔提供支援)。通過新的 Web UI，大家能更好的了解作業的使用情況，記憶體管理也更友善。

1. 本地記憶體（Managed Memory）

Flink 托管記憶體實際上是 Flink 特有的一種本地記憶體，不受 JVM 和 GC 的管理，而是由 Flink 自行進行管理。

本地記憶體的特點主要展現在兩方面：

一方面是 slot 級别的預算規劃，它可以保證作業運作過程中不會因為記憶體不足，造成某些算子或者任務無法運作；也不會因為預留了過多的記憶體沒有使用造成資源浪費。同時 Flink 能保證當任務運作結束時準确将記憶體釋放，確定 Task Manager 執行新任務時有足夠的記憶體可用。
另一方面，資源适應性也是托管記憶體很重要的特性之一，指算子對于記憶體的需求是動态可調整的。具備了适應性，算子就不會因為給予任務過多的記憶體造成資源使用上的浪費，也不會因為提供的記憶體相對較少導緻整個作業無法運作，使記憶體的運用保持在一定的合理範圍内。

當然，在記憶體配置設定相對比較少情況下，作業會受到一定限制，例如需要通過頻繁的落盤保證作業的運作，這樣可能會影響性能。

目前，針對托管記憶體，Flink 的使用場景如下：

RocksDB 狀态後端：在流計算的場景中，每個 Slot 會使用 State 的 Operator，進而共享同一底層的 RocksDB 緩存；
Flink 内置算子：包含批處理、Table SQL、DataSet API 等算子，每個算子有獨立的資源預算，不會互相共享；
Python 程序：使用者使用 PyFlink，使用 Python 語言定義 UDF 時需要啟動 Python 的虛拟機程序。

2. Job Graph 編譯階段

Flink 對于 management memory 的管理主要分為兩個階段。

2.1 作業的 Job Graph 編譯階段

在這個階段需要注意三個問題：

第一個問題是：slot 當中到底有哪些算子或者任務會同時執行。這個問題關系到在一個查詢作業中如何對記憶體進行規劃，是否還有其他的任務需要使用 management memory，進而把相應的記憶體留出來。在流式的作業中，這個問題是比較簡單的，因為我們需要所有的算子同時執行，才能保證上遊産出的資料能被下遊及時的消費掉，這個資料才能夠在整個 job grep 當中流動起來。但是如果我們是在批處理的一些場景當中，實際上我們會存在兩種資料 shuffle 的模式，
- 一種是 pipeline 的模式，這種模式跟流式是一樣的，也就是我們前面說到的 bounded stream 處理方式，同樣需要上遊和下遊的算子同時運作，上遊随時産出，下遊随時消費。
  
  Flink 1.12 資源管理新特性回顧
- 另外一種是所謂的 batch 的 blocking的方式，它要求上遊把資料全部産出，并且落盤結束之後，下遊才能開始讀資料。
這兩種模式會影響到哪些任務可以同時執行。目前在 Flink 當中，根據作業拓撲圖中的一個邊的類型 (如圖上)。我們劃分出了定義的一個概念叫做 pipelined region，也就是全部都由 pipeline 的邊鎖連通起來的一個子圖，我們把這個子圖識别出來，用來判斷哪些 task 會同時執行。
第二個問題是：slot 當中到底有哪些使用場景？我們剛才介紹了三種 manage memory 的使用場景。在這個階段，對于流式作業，可能會出現 Python UDF 以及 Stateful Operator。這個階段當中我們需要注意的是，這裡并不能肯定 State Operator 一定會用到 management memory，因為這跟它的狀态類型是相關的。
- 如果它使用了 RocksDB State Operator，是需要使用 manage memory 的；
- 但是如果它使用的是 Heap State Backend，則并不需要。
然而，作業在編譯的階段，其實并不知道狀态的類型，這裡是需要去注意的地方。
第三個問題：對于 batch 的作業，我們除了需要清楚有哪些使用場景，還需要清楚一件事情，就是前面提到過 batch 的 operator。它使用 management memory 是以一種算子獨享的方式，而不是以 slot 為機關去進行共享。我們需要知道不同的算子應該分别配置設定多少記憶體，這個事情目前是由 Flink 的計劃作業來自動進行設定的。

2.2 執行階段

第一個步驟是根據 State Backend 的類型去判斷是否有 RocksDB。如上圖所示，比如一個 slot，有 ABC 三個算子，B 跟 C 都用到了 Python，C 還用到了 Stateful 的 Operator。這種情況下，如果是在 heap 的情況下，我們走上面的分支，整個 slot 當中隻有一種在使用，就是Python。之後會存在兩種使用方式：

其中一個是 RocksDB State Backend，有了第一步的判斷之後，第二步我們會根據使用者的配置，去決定不同使用方式之間怎麼樣去共享 slot 的 management memory。

在這個 Steaming 的例子當中，我們定義的 Python 的權重是 30%，State Backend 的權重是 70%。在這樣的情況下，如果隻有 Python，Python 的部分自然是使用 100% 的記憶體（Streaming 的 Heap State Backend 分支）；
而對于第二種情況（Streaming 的 RocksDB State Backend 分支），B、C 的這兩個 Operator 共用 30% 的記憶體用于 Python 的 UDF，另外 C 再獨享 70% 的記憶體用于 RocksDB State Backend。最後 Flink 會根據 Task manager 的資源配置，一個 slot 當中有多少 manager memory 來決定每個 operator 實際可以用的記憶體的數量。

批處理的情況跟流的情況有兩個不同的地方，首先它不需要去判斷 State Backend 的類型，這是一個簡化；其次對于 batch 的算子，上文提到每一個算子有自己獨享的資源的預算，這種情況下我們會去根據使用率算出不同的使用場景需要多少的 Shared 之後，還要把比例進一步的細分到每個 Operator。

3. 參數配置

配置參數	預設值	備注
大小	taskmanager.memory.managed.size	/	絕對大小
權重	taskmanager.memory.managed.fraction	0.4	相對大小（占用Flink）總記憶體比例
taskmanager.memory.managed.consumer-weight	DATAPROC:70,PYTHON:30	多種用途并存時候配置設定權重

上方圖表展示了我們需要的是 manager，memory 大小有兩種配置方式：

一種是絕對值的配置方式，
還有一種是作為 Task Manager 總記憶體的一個相對值的配置方式。

taskmanager.memory.managed.consumer-weight 是一個新加的配置項，它的資料類型是 map 的類型，也就是說我們在這裡面實際上是給了一個 key 冒号 value，然後逗号再加上下一組 key 冒号 value 的這樣的一個資料的結構。這裡面我們目前支援兩種 consumer 的 key：

一個是 DATAPROC， DATAPROC 既包含了流處理當中的狀态後端 State Backend 的記憶體，也包含了批處理當中的 Batch Operator；
另外一種是 Python。

二、資源排程

部分資源排程相關的 Feature 是其他版本或者郵件清單裡面大家詢問較多的，這裡我們也做對應的介紹。

1. 最大 Slot 數

Flink 在 1.12 支援了最大 slot 數的一個限制（slotmanager.number-of-slots.max），在之前我們也有提到過對于流式作業我們要求所有的 operator 同時執行起來，才能夠保證資料的順暢的運作。在這種情況下，作業的并發度決定了我們的任務需要多少個 slot 和資源去執行作業。

然而對于批處理其實并不是這樣的，批處理作業往往可以有一個很大的并發度，但實際并不需要這麼多的資源，批處理用很少的資源，跑完前面的任務騰出 Slot 給後續的任務使用。通過這種串行的方式去執行任務能避免 YARN/K8s 叢集的資源過多的占用。目前這個參數支援在 yarn/mesos/native k8 使用。

2. TaskManager 容錯

在我們實際生産中有可能會有程式的錯誤、網絡的抖動、硬體的故障等問題造成 TaskManager 無法連接配接，甚至直接挂掉。我們在日志中常見的就是 TaskManagerLost 這樣的報錯。對于這種情況需要進行作業重新開機，在重新開機的過程中需要重新申請資源和重新開機 TaskManager 程序，這種性能消耗代價是非常高昂的。

對于穩定性要求相對比較高的作業，Flink1.12 提供了一個新的 feature，能夠支援在 Flink 叢集當中始終持有少量的備援的 TaskManager，這些備援的 TaskManager 可以用于在單點故障的時候快速的去恢複，而不需要等待一個重新的資源申請的過程。

通過配置 slotmanager.redundant-taskmanager-num 可以實作備援 TaskManager。這裡所謂的備援 TaskManager 并不是完完全全有兩個 TaskManager 是空負載運作的，而是說相比于我所需要的總共的資源數量，會多出兩個 TaskManager。

任務可能是相對比較均勻的分布在上面，在能夠在利用空閑 TaskManager 的同時，也能夠達到一個相對比較好的負載。一旦發生故障的時候，可以去先把任務快速的排程到現有的還存活的 TaskManager 當中，然後再去進行新一輪的資源申請。目前這個參數支援在 yarn/mesos/native k8 使用。

3. 任務平鋪分布

任務平鋪問題主要出現在 Flink Standalone 模式下或者是比較舊版本的 k8s 模式部署下的。在這種模式下因為事先定義好了有多少個 TaskManager，每個 TaskManager 上有多少 slot，這樣會導緻經常出現排程不均的問題，可能部分 manager 放的任務很滿，有的則放的比較松散。

在 1.11 的版本當中引入了參數 cluster.evenly-spread-out-slots，這樣的參數能夠控制它，去進行一個相對比較均衡的排程。

注意：

第一，這個參數我們隻針對 Standalone 模式，因為在 yarn 跟 k8s 的模式下，實際上是根據你作業的需求來決定起多少 task manager 的，是以是先有了需求再有 TaskManager，而不是先有 task manager，再有 slot 的排程需求。

在每次排程任務的時候，實際上隻能看到目前注冊上來的那一個 TaskManager，Flink 沒辦法全局的知道後面還有多少 TaskManager 會注冊上來，這也是很多人在問的一個問題，就是為什麼特性打開了之後好像并沒有起到一個很好的效果，這是第一件事情。
第二個需要注意的點是，這裡面我們隻能決定每一個 TaskManager 上有多少空閑 slot，然而并不能夠決定每個 operator 有不同的并發數，Flink 并不能決定說每個 operator 是否在 TaskManager 上是一個均勻的分布，因為在 flink 的資源排程邏輯當中，在整個 slot 的 allocation 這一層是完全看不到 task 的。

三、擴充資源架構

1. 背景

近年來，随着人工智能領域的不斷發展，深度學習模型已經被應用到了各種各樣的生産需求中，比較典型的場景如推薦系統，廣告推送，智能風險控制。這些也是 Flink 一直以來被廣泛使用的場景，是以，支援人工智能一直以來都是 Flink 社群的長遠目标之一。針對這個目标，目前已經有了很多第三方的開源擴充工作。由阿裡巴巴開源的工作主要有兩個：

一個是 Flink AI Extended 的項目，是基于 Flink 的深度學習擴充架構，目前支援 TensorFlow、PyTorch 等架構的內建，它使使用者可以将 TensorFlow 當做一個算子，放在 Flink 任務中。
另一個是 Alink，它是一個基于 Flink 的通用算法平台，裡面也内置了很多常用的機器學習算法。

以上的兩個工作都是從功能性上對 Flink 進行擴充，然而從算力的角度上講，深度學習模型亦或機器學習算法，通常都是整個任務的計算瓶頸所在。GPU 則是這個領域被廣泛使用用來加速訓練或者預測的資源。是以，支援 GPU 資源來加速計算是 Flink 在 AI 領域的發展過程中必不可少的功能。

2. 使用擴充資源

目前 Flink 支援使用者配置的資源次元隻有 CPU 與記憶體，而在實際使用中，不僅是 GPU，我們還會遇到其他資源需求，如 SSD 或 RDMA 等網絡加速裝置。是以，我們希望提供一個通用的擴充資源架構，任何擴充資源都可以以插件的形式來加入這個架構，GPU 隻是其中的一種擴充資源。

對于擴充資源的使用，可以抽象出兩個通用需求：

需要支援該類擴充資源的配置與排程。使用者可以在配置中指明對這類擴充資源的需求，如每個 TaskManager 上需要有一塊 GPU 卡，并且當 Flink 被部署在 Kubernetes/Yarn 這類資源底座上時，需要将使用者對擴充資源的需求進行轉發，以保證申請到的 Container/Pod 中存在對應的擴充資源。
需要向算子提供運作時的擴充資源資訊。使用者在自定義算子中可能需要一些運作時的資訊才能使用擴充資源，以 GPU 為例，算子需要知道它内部的模型可以部署在那一塊 GPU 卡上，是以，需要向算子提供這些資訊。

3. 擴充資源架構使用方法

使用資源架構我們可以分為以下這 3 個步驟：

首先為該擴充資源設定相關配置；
然後為所需的擴充資源準備擴充資源架構中的插件；
最後在算子中，從 RuntimeContext 來擷取擴充資源的資訊并使用這些資源

3.1 配置參數

# 定義擴充資源名稱，“gpu”
external-resources: gpu
# 定義每個 TaskManager 所需的 GPU 數量
external-resource.gpu.amount: 1 
# 定義Yarn或Kubernetes中擴充資源的配置鍵
external-resource.gpu.yarn.config-key: yarn.io/gpu
external-resource.gpu.kubernetes.config-key: nvidia.com/gpu
# 定義插件 GPUDriver 的工廠類。
external-resource.gpu.driver-factory.class: 
org.apache.flink.externalresource.gpu.GPUDriverFactory

以上是使用 GPU 資源的配置示例：

對于任何擴充資源，使用者首先需要将它的名稱加入 "external-resources" 中，這個名稱也會被用作該擴充資源其他相關配置的字首來使用。示例中，我們定義了一種名為 "gpu" 的資源。
在排程層，目前支援使用者在 TaskManager 的粒度來配置擴充資源需求。示例中，我們定義每個 TaskManager 上的 GPU 裝置數為 1。
将 Flink 部署在 Kubernetes 或是 Yarn 上時，我們需要配置擴充資源在對應的資源底座上的配置鍵，以便 Flink 對資源需求進行轉發。示例中展示了 GPU 對應的配置。
如果提供了插件，則需要将插件的工廠類名放入配置中。

3.2 前置準備

在實際使用擴充資源前，還需要做一些前置準備工作，以 GPU 為例：

在 Standalone 模式下，叢集管理者需要保證 GPU 資源對 TaskManager 程序可見。
在 Kubernetes 模式下，需要叢集支援 Device Plugin[6]，對應的 Kubernetes 版本為 1.10，并且在叢集中安裝了 GPU 對應的插件。
在 Yarn 模式下，GPU 排程需要叢集 Hadoop 版本在 2.10 或 3.1 以上，并正确配置了 resource-types.xml 等檔案。

3.3 擴充資源架構插件

完成了對擴充資源的排程後，使用者自定義算子可能還需要運作時擴充資源的資訊才能使用它。擴充資源架構中的插件負責完成該資訊的擷取，它的接口如下：

public interface ExternalResourceDriverFactory {
  /**
  * 根據提供的設定建立擴充資源的Driver
  */
  ExternalResourceDriver createExternalResourceDriver(Configuration config) throws Exception;
}

public interface ExternalResourceDriver {
  /**
  * 擷取所需數量的擴充資源資訊
  */
  Set<? extends ExternalResourceInfo> retrieveResourceInfo(long amount) throws Exception;
}

ExternalResourceDriver 會在各個 TaskManager 上啟動，擴充資源架構會調用各個 Driver 的 retrieveResourceInfo 接口來獲得 TaskManager 上的擴充資源資訊，并将得到的資訊傳到算子的 RuntimeContext。ExternalResourceDriverFactory 則為插件的工廠類。

4. GPU 插件

Flink 目前内置了針對 GPU 資源的插件，其内部通過執行名為 Discovery Script 的腳本來擷取目前環境可用的 GPU 資訊，目前資訊中包含了 GPU 裝置的 Index。

Flink 提供了一個預設腳本，位于項目的 "plugins/external-resource-gpu/" 目錄，使用者也可以實作自定義的 Discovery Script 并通過配置來指定使用自定義腳本。該腳本與 GPU 插件的協定為：

當調用腳本時，所需要的 GPU 數量将作為第一個參數輸入，之後為使用者自定義參數清單。
若腳本執行正常，則輸出 GPU Index 清單，以逗号分隔。
若腳本出錯或執行結果不符合預期，則腳本以非零值退出，這會導緻 TaskManager 初始化失敗，并在日志中列印腳本的錯誤資訊。

Flink 提供的預設腳本是通過 "nvidia-smi" 工具來擷取目前的機器中可用的 GPU 數量以及 index，并根據所需要的 GPU 數量傳回對應數量的 GPU Index 清單。當無法擷取到所需數量的 GPU 時，腳本将以非零值退出。

GPU 裝置的資源分為兩個次元，流處理器與顯存，其顯存資源隻支援獨占使用。是以，當多個 TaskManager 運作在同一台機器上時，若一塊 GPU 被多個程序使用，可能導緻其顯存 OOM。是以，Standalone 模式下，需要 TaskManager 級别的資源隔離機制。

預設腳本提供了 Coordination Mode 來支援單機中多個 TaskManager 程序之間的 GPU 資源隔離。該模式通過使用檔案鎖來實作多程序間 GPU 使用資訊同步，協調同一台機器上多個 TaskManager 程序對 GPU 資源的使用。

5. 在算子中擷取擴充資源資訊

在使用者自定義算子中，可使用在 "external-resources" 中定義的資源名稱來調用 RuntimeContext 的 getExternalResourceInfos 接口擷取對應擴充資源的資訊。以 GPU 為例，得到的每個 ExternalResourceInfo 代表一塊 GPU 卡，而其中包含名為 "index" 的字段代表該 GPU 卡的裝置 Index。

public class ExternalResourceMapFunction extends RichMapFunction<String, String> {
  private static finalRESOURCE_NAME="gpu";
  @Override
  public String map(String value) {
    Set<ExternalResourceInfo> gpuInfos = getRuntimeContext().getExternalResourceInfos(RESOURCE_NAME);
    List<String> indexes = gpuInfos.stream()
          .map(gpuInfo -> gpuInfo.getProperty("index").get()).collect(Collectors.toList());
    // Map function with GPU// ...    
  }
}

6. MNIST Demo

下圖以 MNIST 資料集的識别任務來示範使用 GPU 加速 Flink 作業。

MNIST 如上圖所示，為手寫數字圖檔資料集，每個圖檔可表示為為 28*28 的矩陣。在該任務中，我們使用預訓練好的 DNN 模型，圖檔輸入經過一層全連接配接網絡得到一個 10 維向量，該向量最大元素的下标即為識别結果。

我們在一台擁有兩塊 GPU 卡的 ECS 上啟動一個有兩個 TaskManager 程序的 Standalone 叢集。借助預設腳本提供的 Coordination Mode 功能，我們可以保證每個 TaskManager 各使用其中一塊 GPU 卡。

該任務的核心算子為圖像識别函數 MNISTClassifier，核心實作如下所示

class MNISTClassifier extends RichMapFunction<List<Float>, Integer> {

  @Override
  public void open(Configuration parameters) {
    //擷取GPU資訊并且選擇第一塊GPU
    Set<ExternalResourceInfo> externalResourceInfos =   getRuntimeContext().getExternalResourceInfos(resourceName);
    final Optional<String> firstIndexOptional = externalResourceInfos.iterator().next().getProperty("index");
    // 使用第一塊GPU的index初始化JCUDA元件
    JCuda.cudaSetDevice(Integer.parseInt(firstIndexOptional.get()));
    JCublas.cublasInit();
  }
}

在 Open 方法中，從 RuntimeContext 擷取目前 TaskManager 可用的 GPU，并選擇第一塊來初始化 JCuda 以及 JCublas 庫。

class MNISTClassifier extends RichMapFunction<List<Float>, Integer> {
    @Override
    public Integer map(List<Float> value) {
        // 使用Jucblas做矩陣算法
        JCublas.cublasSgemv('n', DIMENSIONS.f1, DIMENSIONS.f0, 1.0f,
                matrixPointer, DIMENSIONS.f1, inputPointer, 1, 0.0f, outputPointer, 1);

        // 獲得乘法結果并得出該圖所表示的數字
        JCublas.cublasGetVector(DIMENSIONS.f1, Sizeof.FLOAT, outputPointer, 1, Pointer.to(output), 1);

        JCublas.cublasFree(inputPointer);
        JCublas.cublasFree(outputPointer);

        int result = 0;
        for (int i = 0; i < DIMENSIONS.f1; ++i) {
            result = output[i] > output[result] ? i : result;
        }
        return result;
    }
}

在 Map 方法中，将預先訓練好的模型參數與輸入矩陣放入 GPU 顯存，使用 JCublas 進行 GPU 中的矩陣乘法運算，最後将結果向量從 GPU 顯存中取出并得到識别結果數字。

具體案例示範流程可以前往觀看視訊或者參考 Github 上面的連結動手嘗試。

四、未來計劃

除了上文介紹的這些已經釋出的特性外，Apache Flink 社群也正在積極準備更多資源管理方面的優化特性，在未來的版本中将陸續和大家見面。

被動資源排程模式：托管記憶體使得 Flink 任務可以靈活地适配不同的 TaskManager/Slot 資源，充分利用可用資源，為計算任務提供給定資源限制下的最佳算力。但使用者仍需指定計算任務的并行度，Flink 需要申請到滿足該并行度數量的 TaskManager/Slot 才能順利執行。被動資源排程将使 Flink 能夠根據可用資源動态改變并行度，在資源不足時能夠 best effort 進行資料處理，同時在資源充足時恢複到指定的并行度保障處理性能。
細粒度資源管理：Flink 目前基于 Slot 的資源管理與排程機制，認為所有的 Slot 都具有相同的規格。對于一些複雜的規模化生産任務，往往需要将計算任務拆分成多個子圖，每個子圖單獨使用一個 Slot 執行。當子圖間的資源需求差異較大時，使用相同規格的 Slot 往往難以滿足資源效率方面的需求，特别是對于 GPU 這類成本較高的擴充資源。細粒度資源管理允許使用者為作業的子圖指定資源需求，Flink 會根據資源需求使用不同規格的 TaskManager/Slot 執行計算任務，進而優化資源效率。

五、總結

通過文章的介紹，相信大家對 Flink 記憶體管理有了更加清晰的認知。

首先從本地記憶體、Job Graph 編譯階段、執行階段來解答每個流程的記憶體管理以及記憶體配置設定細節，通過新的參數配置控制 TaskManager的記憶體配置設定；
然後從大家平時遇到資源排程相關問題，包括最大 Slot 數使用，如何進行 TaskManager 進行容錯，任務如何通過任務平鋪均攤任務資源；
最後在機器學習和深度學習領域常常用到 GPU 進行加速計算，通過解釋 Flink 在 1.12 版本如何使用擴充資源架構和示範 Demo，給我們展示了資源擴充的使用。再針對資源使用率方面提出 2 個社群未來正在做的計劃，包括被動資源模式和細粒度的資源管理。

六、附錄

[1]

Accelerating your workload with GPU and other external resources

[2]

擴充資源架構文檔

[3]

FLIP-108: Add GPU support in Flink

[4]

flink-mnist 項目

更多 Flink 相關技術問題，可掃碼加入社群釘釘交流群～

活動推薦

阿裡雲基于 Apache Flink 建構的企業級産品-實時計算Flink版現開啟活動：

99元試用

實時計算Flink版

(包年包月、10CU）即有機會獲得 Flink 獨家定制T恤；另包3個月及以上還有85折優惠！

了解活動詳情：

https://www.aliyun.com/product/bigdata/sc