[[410005]]

本文轉(zhuǎn)載自微信公眾號(hào)「肌肉碼農(nóng)」，作者鄒學(xué)。轉(zhuǎn)載本文請(qǐng)聯(lián)系肌肉碼農(nóng)公眾號(hào)。

插件名稱：flink-connector-redis

插件地址：https://github.com/jeff-zou/flink-connector-redis.git

項(xiàng)目介紹

基于bahir-flink二次開(kāi)發(fā)，使它支持SQL直接定義寫(xiě)入redis,用戶通過(guò)DDL指定自己需要保存的字段。

使用方法:

命令行執(zhí)行 mvn package -DskipTests=true打包后，將生成的包flink-connector-redis_2.12-1.11.1.jar引入flink lib中即可，無(wú)需其它設(shè)置。

重構(gòu)介紹：

相對(duì)上一個(gè)版本簡(jiǎn)化了參數(shù)設(shè)置，思路更清晰，上一版本字段的值會(huì)根據(jù)主鍵等條件來(lái)自動(dòng)生成，這要求使用者需要了解相關(guān)規(guī)則，有一定的學(xué)習(xí)成本并且容易埋坑，重構(gòu)后字段的值由用戶在DDL中顯示地指定，如下：

'key-column'='username','value-column'='passport',' //直接指定字段名 

取消了必須有主鍵的限制，使用更簡(jiǎn)單，如果有多個(gè)字段組合成key或者value,需要用戶在DML中使用concat_ws等方式組裝，不再是插件在后臺(tái)用不可見(jiàn)字符拼裝。

使用示例:

• 1.SQL方式

示例代碼路徑: src/test/java/org.apache.flink.streaming.connectors.redis.table.SQLInsertTest.java

set示例，相當(dāng)于redis命令： set test test11

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build(); 
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, environmentSettings); 
 
        String ddl = "create table sink_redis(username VARCHAR, passport VARCHAR) with ( 'connector'='redis', " + 
                "'host'='10.11.80.147','port'='7001', 'redis-mode'='single','password'='******','key-column'='username','value-column'='passport','command'='set')" ; 
 
        tEnv.executeSql(ddl); 
        String sql = " insert into sink_redis select * from (values ('test', 'test11'))"; 
        TableResult tableResult = tEnv.executeSql(sql); 
        tableResult.getJobClient().get() 
                .getJobExecutionResult() 
                .get(); 

• 2.DataStream方式

示例代碼路徑:

src/test/java/org.apache.flink.streaming.connectors.redis.datastream.DataStreamInsertTest.java

hset示例，相當(dāng)于redis命令：hset tom math 150

Configuration configuration = new Configuration(); 
        configuration.setString(RedisOptions.KEY_COLUMN, "name"); 
        configuration.setString(RedisOptions.FIELD_COLUMN, "subject"); //對(duì)應(yīng)hash的field、 sorted set的score 
        configuration.setString(RedisOptions.VALUE_COLUMN, "score"); 
        configuration.setString(REDIS_MODE, REDIS_CLUSTER); 
        configuration.setString(REDIS_COMMAND, RedisCommand.HSET.name()); 
 
        RedisMapper redisMapper = RedisHandlerServices 
                .findRedisHandler(RedisMapperHandler.class, configuration.toMap()) 
                .createRedisMapper(configuration); 
 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
 
        GenericRowData genericRowData = new GenericRowData(3); 
        genericRowData.setField(0, "tom"); 
        genericRowData.setField(1, "math"); 
        genericRowData.setField(2, "150"); 
        DataStream<GenericRowData> dataStream = env.fromElements(genericRowData); 
 
        TableSchema tableSchema =  new TableSchema.Builder() .field("name", DataTypes.STRING().notNull()).field("subject", DataTypes.STRING()).field("score", DataTypes.INT()).build(); 
 
        FlinkJedisConfigBase conf = getLocalRedisClusterConfig(); 
        RedisSink redisSink = new RedisSink<>(conf, redisMapper, tableSchema); 
 
        dataStream.addSink(redisSink); 
        env.execute("RedisSinkTest");