Flink1.12 SQL向Redis实时写数据

Flink1.12 SQL向Redis实时写数据

插件名称：flink-connector-redis

项目介绍

基于bahir-flink二次开发，使它支持SQL直接定义写入redis,用户通过DDL指定自己需要保存的字段。

使用方法:

命令行执行 mvn package -DskipTests=true打包后，将生成的包flink-connector-redis_2.12-1.11.1.jar引入flink lib中即可，无需其它设置。

重构介绍：

相对上一个版本简化了参数设置，思路更清晰，上一版本字段的值会根据主键等条件来自动生成，这要求使用者需要了解相关规则，有一定的学习成本并且容易埋坑，重构后字段的值由用户在DDL中显示地指定，如下：

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

取消了必须有主键的限制，使用更简单，如果有多个字段组合成key或者value,需要用户在DML中使用concat_ws等方式组装，不再是插件在后台用不可见字符拼装。

使用示例:

1.SQL方式

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

set示例，相当于redis命令： set test test11

2.DataStream方式

示例代码路径:

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

hset示例，相当于redis命令：hset tom math 150

Configuration configuration = new Configuration();         configuration.setString(RedisOptions.KEY_COLUMN, "name");         configuration.setString(RedisOptions.FIELD_COLUMN, "subject"); //对应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 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");

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。