開源組件Flink關于redisSink業務需求改造分享

IT那活兒發布于2023-01-11 13:20 / 2744人閱讀

[

背景

]

由于項目需求，用Flink生成完實時數據之后，需要立即給前臺做展示，且日志項目實時數據數量較大，因此采用了Redis作為數據庫，用來存儲最近一小段時間的數據，供前臺來進行實時展示。

[

問題

]

Flink的Sink方法提供了一個redis的sink包（flink-connector-redis_2.11），但是使用過程中發現與業務不符合。主要是2個方面：

Flink提供的包入redis時不能設置失效時間，業務場景是只需要保存最新的實時數據一段時間，需要設置失效時間，讓數據過期，從而提高資源的利用和性能。
業務使用的是Redis集群模式，Flink提供的Jar包雖然支持集群但是沒有設置密碼的地方，業務的Redis集群都有設置密碼的強制要求。

[

解決方案

]

對Flink提供RedisSink包進行改造，具體的是：1.增加可以設置失效時間的方法；2.集群模式提供Redis的驗證方法。

1、增加缺失的方法

由于原來的RedisMapper只提供了獲取入庫方式、獲取key以及獲取Value，三個接口，沒有提供獲取失效時間的接口。

所以首先重新定義RedisMapper接口類，增加一個獲取失效時間的接口。

其次在RedisCommand中增加一個帶失效時間的命令

SETEX(BasicRedisDataType.STRING)接口既然以及改好了，就要去重新定義它的實現了。

接著在RedisSink的invoke方法中提供獲取失效時間的代碼，以及SETNX命令的實現。

由于原來的RedisCommandsContainer接口中不包含失效時間的方法，所以需要新增一個含失效時間的接口。

在具體的RedisClusterContainer集群實現該方法。

這樣第一個問題就解決了，只要將設置為RedisCommand.SETEX就可以進行帶失效時間的Sink方法了。

2、增加缺失的參數

關于增加設置密碼的方式，Flink提供的包中JedisClusterConfig是缺失了集群密碼設置的。因此需要添加一個密碼選項，并提供set方法。

然后初始化的時候增加一個構造方法。

其實最終還是調用的原生redis連接包來創建了一個集群對象redis.clients.jedis.JedisCluster，在RedisCommandsContainerBuilder中添加帶密碼的redis集群創建方式。

通過這樣改造就擁有了設置密碼的方式，如果還少了別的參數同理可以通過這樣的方式給添加上去。為了靈活改造我們實際直接將整個jar拉取了下來，重新定義了一個屬于自己的Sink定制包，最終使用如下：

簡要的給出類圖

[

總結

]

本文主要通過結合項目的實際使用場景，在flink與redis的sink包（flink-connector-redis_2.11）不能滿足要求的情況下，在原有包的基礎上根據實際使用情況，對該包進行了一系列的改造，主要包括提供設置密碼的redis集群初始化以及在sink過程中可以指定redis鍵值的失效時長，最終通過改造的flink-connector-redis在項目flink任務中使用良好。

云服務器 GPU云服務器分享組件關于服務器的需求分析 Android開源組件開源bug需求管理工具

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/130151.html

Apache Flink，流計算？不僅僅是流計算！

摘要：基于流處理機制實現批流融合相對基于批處理機制實現批流融合的思想更自然，更合理，也更有優勢，因此阿里巴巴在基于支持大量核心實時計算場景的同時，也在不斷改進的架構，使其朝著真正批流融合的統一計算引擎方向前進。阿里妹導讀：2018年12月下旬，由阿里巴巴集團主辦的Flink Forward China在北京國家會議中心舉行。Flink Forward是由Apache軟件基金會授權的全球范圍...

KoreyLee 2019-06-24 18:16 評論0 收藏0
取之開源，用之開源——深度剖析阿里巴巴對Apache Flink的優化與改進

摘要：基于在阿里巴巴搭建的平臺于年正式上線，并從阿里巴巴的搜索和推薦這兩大場景開始實現。在經過一番調研之后，阿里巴巴實時計算認為是一個非常適合的選擇。接下來，我們聊聊阿里巴巴在層對又大刀闊斧地進行了哪些改進。 Apache Flink 概述 Apache Flink（以下簡稱Flink）是誕生于歐洲的一個大數據研究項目，原名StratoSphere。該項目是柏林工業大學的一個研究性項目，早期...

YJNldm 2019-07-24 10:49 評論0 收藏0
OPPO數據中臺之基石：基于Flink SQL構建實數據倉庫

摘要：實際上，本身就預留了與外部元數據對接的能力，分別提供了和這兩個抽象。對接外部數據源搞清楚了注冊庫表的過程，給我們帶來這樣一個思路如果外部元數據創建的表也能被轉換成可識別的，那么就能被無縫地注冊到。本文整理自 2019 年 4 月 13 日在深圳舉行的 Flink Meetup 會議，分享嘉賓張俊，目前擔任 OPPO 大數據平臺研發負責人，也是 Apache Flink contrib...

jeffrey_up 2019-07-24 10:55 評論0 收藏0