當前位置：首頁 > 运维知识 > windows >内容正文

windows

Flink异步IO

發布時間：2024/1/16 windows 62 coder

生活随笔收集整理的這篇文章主要介紹了 Flink异步IO 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

本文講解 Flink 用于訪問外部數據存儲的異步 I/O API。對于不熟悉異步或者事件驅動編程的用戶，建議先儲備一些關于 Future 和事件驅動編程的知識。

對于異步 I/O 操作的需求

在與外部系統交互（用數據庫中的數據擴充流數據）的時候，需要考慮與外部系統的通信延遲對整個流處理應用的影響。

簡單地訪問外部數據庫的數據，比如使用 MapFunction，通常意味著同步交互： MapFunction 向數據庫發送一個請求然后一直等待，直到收到響應。在許多情況下，等待占據了函數運行的大部分時間。

與數據庫異步交互是指一個并行函數實例可以并發地處理多個請求和接收多個響應。這樣，函數在等待的時間可以發送其他請求和接收其他響應。至少等待的時間可以被多個請求攤分。大多數情況下，異步交互可以大幅度提高流處理的吞吐量。

注意：僅僅提高 MapFunction 的并行度（parallelism）在有些情況下也可以提升吞吐量，但是這樣做通常會導致非常高的資源消耗：更多的并行 MapFunction 實例意味著更多的 Task、更多的線程、更多的 Flink 內部網絡連接、更多的與數據庫的網絡連接、更多的緩沖和更多程序內部協調的開銷。

先決條件

如上節所述，正確地實現數據庫（或鍵/值存儲）的異步 I/O 交互需要支持異步請求的數據庫客戶端。許多主流數據庫都提供了這樣的客戶端。

如果沒有這樣的客戶端，可以通過創建多個客戶端并使用線程池處理同步調用的方法，將同步客戶端轉換為有限并發的客戶端。然而，這種方法通常比正規的異步客戶端效率低。

異步 I/O API

Flink 的異步 I/O API 允許用戶在流處理中使用異步請求客戶端。API 處理與數據流的集成，同時還能處理好順序、事件時間和容錯等。

在具備異步數據庫客戶端的基礎上，實現數據流轉換操作與數據庫的異步 I/O 交互需要以下三部分：

實現分發請求的 AsyncFunction
獲取數據庫交互的結果并發送給 ResultFuture 的回調函數
將異步 I/O 操作應用于 DataStream 作為 DataStream 的一次轉換操作, 啟用或者不啟用重試。
下面是基本的代碼模板：

// 這個例子使用 Java 8 的 Future 接口（與 Flink 的 Future 相同）實現了異步請求和回調。

/**
 * 實現 'AsyncFunction' 用于發送請求和設置回調。
 */
class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {

    /** 能夠利用回調函數并發發送請求的數據庫客戶端 */
    private transient DatabaseClient client;

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new DatabaseClient(host, post, credentials);
    }

    @Override
    public void close() throws Exception {
        client.close();
    }

    @Override
    public void asyncInvoke(String key, final ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {

        // 發送異步請求，接收 future 結果
        final Future<String> result = client.query(key);

        // 設置客戶端完成請求后要執行的回調函數
        // 回調函數只是簡單地把結果發給 future
        CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override
            public String get() {
                try {
                    return result.get();
                } catch (InterruptedException | ExecutionException e) {
                    // 顯示地處理異常。
                    return null;
                }
            }
        }).thenAccept( (String dbResult) -> {
            resultFuture.complete(Collections.singleton(new Tuple2<>(key, dbResult)));
        });
    }
}

// 創建初始 DataStream
DataStream<String> stream = ...;

// 應用異步 I/O 轉換操作，不啟用重試
DataStream<Tuple2<String, String>> resultStream =
    AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);

這是官網的代碼模板，這里給出Flink異步IO訪問mysql數據的例子。

數據庫有一張people表，字段姓名和國家，Flink從nc讀取數據，根據空格切分人名，從mysql查出每個人對應的國家，然后打印出來。真實大數據場景可能會遇到其它的外部存儲，需要在Flink程序里面訪問這些數據庫，擴充數據維度，組成大寬表。比如redis、hbase等數據庫。

mysql里面提前建好people表，建表語句：

CREATE TABLE `people` (
  `id` bigint(13) unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(50) NOT NULL,
  `country` varchar(50) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of people
-- ----------------------------
INSERT INTO `people` VALUES ('1', 'tom', 'US');
INSERT INTO `people` VALUES ('2', 'zhangsan', 'china');

我們的程序：

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid</artifactId>
    <version>1.2.8</version>
</dependency>

package operator;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.*;
import java.util.Collections;
import java.util.concurrent.*;
import java.util.function.Supplier;

public class AsyncDataBaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {
    // 線程池
    private ExecutorService executorService;

    // 連接池
    private DruidDataSource druidDataSource;

    @Override
    public void asyncInvoke(String key, ResultFuture<Tuple2<String, String>> resultFuture) {
        Future<String> result = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                // 從連接池中獲取連接
                DruidPooledConnection connection = druidDataSource.getConnection();
                // 預編譯SQL
                String sql = "select country from people where name = ?";
                PreparedStatement preparedStatement = connection.prepareStatement(sql);
                // 設置參數
                preparedStatement.setString(1, key);
                // 執行SQL并獲取結果
                ResultSet resultSet = preparedStatement.executeQuery();

                String country = "";
                try {
                    // 封裝結果
                    while (resultSet.next()) {
                        country = resultSet.getString("country");
                    }
                } finally {
                    resultSet.close();
                    preparedStatement.close();
                    connection.close();
                }
                return country;
            }
        });

        // 獲取異步結果并輸出
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    return result.get();
                } catch (InterruptedException | ExecutionException e) {
                    return null;
                }
            }
        }).thenAccept((String dbResult) -> {
            resultFuture.complete(Collections.singleton(Tuple2.of(key, dbResult)));
        });

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        druidDataSource = new DruidDataSource();
        druidDataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        druidDataSource.setUsername("bigdata");
        druidDataSource.setPassword("bigdata");
        druidDataSource.setUrl("jdbc:mysql://192.168.1.1:3306/test");

        // 創建線程池，用于執行異步操作
        executorService = new ThreadPoolExecutor(5, 15, 1,
                TimeUnit.MINUTES,
                new LinkedBlockingDeque<>(100));
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 關閉連接池
        if (druidDataSource != null){
            druidDataSource.close();
        }

        // 關閉線程池
        if (executorService != null){
            executorService.shutdown();
        }
    }
}

主程序：

package operator;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class AsyncIODemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> stream = senv.socketTextStream("192.168.20.130", 9999)
                .flatMap(new FlatMapFunction<String, String>() {

                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        String[] values = value.split(" ");
                        for(String v : values) {
                            out.collect(v);
                        }
                    }
                });
        // 應用異步 I/O 轉換操作，不啟用重試
        DataStream<Tuple2<String, String>> resultStream =
                AsyncDataStream.unorderedWait(stream, new AsyncDataBaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);

        resultStream.print();

        senv.execute("AsyncIODemo");

    }
}

啟動程序，nc輸入數據：

[root@hm-001 logs]# nc -lk 9999
tom zhangsan tom

程序輸出：

6> (zhangsan,china)
6> (tom,US)
6> (tom,US)

代碼gitee地址：

https://gitee.com/ddxygq/BigDataTechnical/blob/main/Flink/src/main/java/operator/AsyncIODemo.java

重要提示：第一次調用 ResultFuture.complete 后 ResultFuture 就完成了。后續的 complete 調用都將被忽略。

下面兩個參數控制異步操作：

Timeout：超時參數定義了異步操作執行多久未完成、最終認定為失敗的時長，如果啟用重試，則可能包括多個重試請求。它可以防止一直等待得不到響應的請求。
Capacity：容量參數定義了可以同時進行的異步請求數。即使異步 I/O 通常帶來更高的吞吐量，執行異步 I/O 操作的算子仍然可能成為流處理的瓶頸。限制并發請求的數量可以確保算子不會持續累積待處理的請求進而造成積壓，而是在容量耗盡時觸發反壓。
AsyncRetryStrategy: 重試策略參數定義了什么條件會觸發延遲重試以及延遲的策略，例如，固定延遲、指數后退延遲、自定義實現等。

超時處理

當異步 I/O 請求超時的時候，默認會拋出異常并重啟作業。如果你想處理超時，可以重寫 AsyncFunction#timeout 方法。重寫 AsyncFunction#timeout 時別忘了調用 ResultFuture.complete() 或者 ResultFuture.completeExceptionally() 以便告訴Flink這條記錄的處理已經完成。如果超時發生時你不想發出任何記錄，你可以調用 ResultFuture.complete(Collections.emptyList()) 。

結果的順序

AsyncFunction 發出的并發請求經常以不確定的順序完成，這取決于請求得到響應的順序。 Flink 提供兩種模式控制結果記錄以何種順序發出。

無序模式：異步請求一結束就立刻發出結果記錄。流中記錄的順序在經過異步 I/O 算子之后發生了改變。當使用處理時間作為基本時間特征時，這個模式具有最低的延遲和最少的開銷。此模式使用 AsyncDataStream.unorderedWait(...) 方法。
有序模式: 這種模式保持了流的順序。發出結果記錄的順序與觸發異步請求的順序（記錄輸入算子的順序）相同。為了實現這一點，算子將緩沖一個結果記錄直到這條記錄前面的所有記錄都發出（或超時）。由于記錄或者結果要在 checkpoint 的狀態中保存更長的時間，所以與無序模式相比，有序模式通常會帶來一些額外的延遲和 checkpoint 開銷。此模式使用 AsyncDataStream.orderedWait(...) 方法。

事件時間

當流處理應用使用事件時間時，異步 I/O 算子會正確處理 watermark。對于兩種順序模式，這意味著以下內容：

無序模式： Watermark 既不超前于記錄也不落后于記錄，即 watermark 建立了順序的邊界。只有連續兩個 watermark 之間的記錄是無序發出的。在一個 watermark 后面生成的記錄只會在這個 watermark 發出以后才發出。在一個 watermark 之前的所有輸入的結果記錄全部發出以后，才會發出這個 watermark。這意味著存在 watermark 的情況下，無序模式會引入一些與有序模式相同的延遲和管理開銷。開銷大小取決于 watermark 的頻率。
有序模式：連續兩個 watermark 之間的記錄順序也被保留了。開銷與使用處理時間相比，沒有顯著的差別。

請記住，攝入時間是一種特殊的事件時間，它基于數據源的處理時間自動生成 watermark。

容錯保證

異步 I/O 算子提供了完全的精確一次容錯保證。它將在途的異步請求的記錄保存在 checkpoint 中，在故障恢復時重新觸發請求。

重試支持

重試支持為異步 I/O 操作引入了一個內置重試機制，它對用戶的異步函數實現邏輯是透明的。

AsyncRetryStrategy: 異步重試策略包含了觸發重試條件 AsyncRetryPredicate 定義，以及根據當前已嘗試次數判斷是否繼續重試、下次重試間隔時長的接口方法。需要注意，在滿足觸發重試條件后，有可能因為當前重試次數超過預設的上限放棄重試，或是在任務結束時被強制終止重試（這種情況下，系統以最后一次執行的結果或異常作為最終狀態）。
AsyncRetryPredicate: 觸發重試條件可以選擇基于返回結果、執行異常來定義條件，兩種條件是或的關系，滿足其一即會觸發。

實現提示

在實現使用 Executor（或者 Scala 中的 ExecutionContext）和回調的 Futures 時，建議使用 DirectExecutor，因為通常回調的工作量很小，DirectExecutor 避免了額外的線程切換開銷。回調通常只是把結果發送給 ResultFuture，也就是把它添加進輸出緩沖。從這里開始，包括發送記錄和與 chenkpoint 交互在內的繁重邏輯都將在專有的線程池中進行處理。

DirectExecutor 可以通過 org.apache.flink.util.concurrent.Executors.directExecutor() 或 com.google.common.util.concurrent.MoreExecutors.directExecutor() 獲得。

總結

以上是生活随笔為你收集整理的Flink异步IO的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

io
Flink

上一篇： Sunshine + Moonlight
下一篇： Apollo系列之架构设计（一）