Flink從socket讀取數據sink到redis

import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.redis.RedisSink; import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig; import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand; import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription; import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper; /** * Flink從socket讀取數據sink到redis * */
public class MyRedisSink { public static void main(String[] args) throws Exception { // 獲取一個執行環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //安裝nc命令：yum install -y nc //nc -l 33069 ##執行該命令，然后一行一行輸入字符
        DataStreamSource<String> dataStream = env.socketTextStream("106.12.241.89", 33069, "\n"); //lpsuh reids_words => socket發送過來的數據，最終會存儲到redis中//對數據進行組裝,把string轉化為tuple2<String,String>
        DataStream<Tuple2<String, String>> redis_wordsData = dataStream .map(new MapFunction<String, Tuple2<String, String>>() { @Override public Tuple2<String, String> map(String s) throws Exception { System.out.println("socket發送過來的數據： " + s); return new Tuple2<>("reids_words", s); } }); //創建redis的配置
        FlinkJedisPoolConfig build = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setPort(6379).build(); //創建redissink
        RedisSink<Tuple2<String, String>> redisSink = new RedisSink<>(build, new MyRedisMapper()); redis_wordsData.addSink(redisSink); env.execute("MyRedisSink"); } public static class MyRedisMapper implements RedisMapper<Tuple2<String, String>> { /** * 表示從接收的數據中獲取需要操作的redis key */ @Override public String getKeyFromData(Tuple2<String, String> data) { return data.f0; //第一個元素
 } /** * 表示從接收的數據中獲取需要操作的redis value */ @Override public String getValueFromData(Tuple2<String, String> data) { return data.f1; //第二個元素
 } @Override public RedisCommandDescription getCommandDescription() { return new RedisCommandDescription(RedisCommand.LPUSH); } } }

 

通過nc命令模擬發送數據：

 

控制台打印的信息：

socket發送過來的數據： aa socket發送過來的數據： bb socket發送過來的數據： cc socket發送過來的數據： dd socket發送過來的數據： 11 socket發送過來的數據： 22

 

在Redis中查看socket發送過來的數據：