图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Flink › 技术应用› flink1.12自定义redis sink
分享

flink1.12自定义redis sink

Mirinda 2021-3-25 10:38:10 发表于 技术应用 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 7022
问题导读:
1.代码从11升级到了12本文举例废弃了什么方法？
2.如何自定义的 sink类实现 DynamicTableSink 接口？
3.getChangelogMode设置什么？

一.变更简述:
最近代码从11升级到了12，由于此次版本变动，废弃了很多api，导致代码不兼容。如原来的ta b leEnv中注册registerTablesink和registerTableSource方法


a1.png 

  1.  /**     * Registers an external {@link TableSink} with given field names and types in this
复制代码
说明中要求通过 executeSql(ddl) 的方法注册一个表。
既然要通过ddl创建，那就要自定义一个redis 的connector。
flink自定的connector是借助 TableFactory spi发现注册的。
所以一共需要实现两个重要的类。


a2.png


a3.png

二 .关键类介绍
1.RedisDynamicTableSourceFactory类
自定义的facory类只需要实现 DynamicTableSinkFactory 接口即可。
  1. import com.iqiyi.talos.engine.operators.sink.dynamic.RedisDynamicTableSink;

  2. import org.apache.flink.configuration.ConfigOption;

  3. import org.apache.flink.configuration.ReadableConfig;

  4. import org.apache.flink.table.connector.sink.DynamicTableSink;

  5. import org.apache.flink.table.factories.DynamicTableSinkFactory;

  6. import org.apache.flink.table.factories.FactoryUtil;

  7. 

  8. import java.util.HashSet;

  9. import java.util.Set;

  10. 

  11. /**

  12.  * @ClassName RedisDynamicTableSourceFactory

  13.  * @Description 自定义RedisSinkFactory

  14.  * @Author zwc

  15.  * @Date 2021-01-11 11:44

  16.  * @Version 1.0

  17.  **/

  18. public class RedisDynamicTableSourceFactory implements DynamicTableSinkFactory {

  19. //    public static final ConfigOption<String> port = ConfigOptions.key("host").stringType().noDefaultValue();

  20. 

  21. 

  22.     @Override

  23.     public DynamicTableSink createDynamicTableSink(Context context) {

  24.         final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);

  25.         helper.validate();

  26.         ReadableConfig options = helper.getOptions();

  27.         return new RedisDynamicTableSink(options);

  28.     }

  29.     //connector = "redis" 声明连接器名称

  30.     @Override

  31.     public String factoryIdentifier() {

  32.         return "redis";

  33.     }

  34. 

  35.     @Override

  36.     public Set<ConfigOption<?>> requiredOptions() {

  37.         Set<ConfigOption<?>> options = new HashSet();

  38.         return options;

  39.     }

  40. 

  41.     @Override

  42.     public Set<ConfigOption<?>> optionalOptions() {

  43.         Set<ConfigOption<?>> options = new HashSet();

  44. //        options.add(port);

  45.         return options;

  46.     }

  47. }
复制代码
createDynamicTableSink: 校验ddl中 with(...) 附加的选项，并且从CatalogTable 初始化实例，将Options附加选项加载到上下文Context中。
factoryIdentifier: 连接器的名称
requiredOptions：必填参数
optionalOptions：可选参数

2.RedisDynamicTableSink类
自定义的 sink类实现 DynamicTableSink 接口
  1. import com.iqiyi.talos.engine.job.EngineContext;

  2. import com.iqiyi.talos.engine.operators.function.CollectionTableSinkFunction;

  3. import org.apache.flink.configuration.ReadableConfig;

  4. import org.apache.flink.table.connector.ChangelogMode;

  5. import org.apache.flink.table.connector.sink.DynamicTableSink;

  6. import org.apache.flink.table.connector.sink.SinkFunctionProvider;

  7. 

  8. /**

  9.  * @ClassName RedisDynamicTableSink

  10.  * @Description TODO

  11.  * @Author zwc

  12.  * @Date 2021-01-11 15:41

  13.  * @Version 1.0

  14.  **/

  15. public class RedisDynamicTableSink implements DynamicTableSink {

  16. 

  17.     private ReadableConfig options;

  18. 

  19.     private EngineContext ctx;

  20. 

  21.     public RedisDynamicTableSink(ReadableConfig options) {

  22.         this.options = options;

  23.         ctx = EngineContext.getContext();

  24.     }

  25. 

  26. 

  27.     @Override

  28.     public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {

  29.         return ChangelogMode.insertOnly();

  30.     }

  31. 

  32.     @Override

  33.     public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {

  34. 

  35.         CollectionTableSinkFunction collectionTableSinkFunction = new CollectionTableSinkFunction(ctx.getDeploySite(), ctx.getSimpifiedName());

  36.         return SinkFunctionProvider.of(collectionTableSinkFunction);

  37.     }

  38. 

  39.     @Override

  40.     public DynamicTableSink copy() {

  41.         return new RedisDynamicTableSink(this.options);

  42.     }

  43. 

  44.     @Override

  45.     public String asSummaryString() {

  46.         return "my_redis_sink";

  47.     }

  48. }
复制代码
getChangelogMode: 设置sink 是 insert only模式（目前只支持这种模式）
getSinkRuntimeProvider: 这里就是执行sink的具体逻辑了。这里可以直接使用flink 提供的redis-connector
  1. <dependency>

  2.             <groupId>org.apache.bahir</groupId>

  3.             <artifactId>flink-connector-redis_2.11</artifactId>

  4.             <version>1.0</version>

  5.         </dependency>
复制代码
  1. RedisMapper<RowData> stringRedisMapper = new RedisMapper<RowData>() {

  2.             //返回对应Redis命令

  3.             @Override

  4.             public RedisCommandDescription getCommandDescription() {

  5.                 return new RedisCommandDescription(RedisCommand.SET);

  6.             }

  7.             //从数据中获取对应Key

  8.             @Override

  9.             public String getKeyFromData(RowData rowData) {

  10.                 StringData string = rowData.getString(0);

  11.                 return string;

  12.             }

  13.             //从数据中获取对应Value

  14.             @Override

  15.             public String getValueFromData(RowData rowData) {

  16.                 String s = rowData.toString();

  17.                 return s;
复制代码
由于此处我需要自定义redis数据结构，所以自己实现了RichFunction
  1. @Override

  2.     public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {

  3. 

  4.         CollectionTableSinkFunction collectionTableSinkFunction = new CollectionTableSinkFunction(ctx.getDeploySite(), ctx.getSimpifiedName());

  5.         return SinkFunctionProvider.of(collectionTableSinkFunction);

  6.     }
复制代码
  1. /**

  2.  * @ClassName CollectionTableSinkFunction

  3.  * @Description TODO

  4.  * @Author zwc

  5.  * @Date 2021-01-12 16:51

  6.  * @Version 1.0

  7.  **/

  8. 

  9. import com.alibaba.fastjson.JSON;

  10. import com.iqiyi.talos.common.JedisClient;

  11. import com.iqiyi.talos.engine.enums.DeploySite;

  12. import org.apache.flink.configuration.Configuration;

  13. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

  14. import org.apache.flink.table.data.RowData;

  15. import org.slf4j.Logger;

  16. import org.slf4j.LoggerFactory;

  17. 

  18. import java.util.ArrayList;

  19. import java.util.List;

  20. import java.util.Map;

  21. import java.util.concurrent.ConcurrentHashMap;

  22. 

  23. /**

  24.  * 这里定义了当接收到一条数据时，该如何 sink 的具体逻辑

  25.  */

  26. public class CollectionTableSinkFunction extends RichSinkFunction<RowData> {

  27.     private static Logger LOG = LoggerFactory.getLogger(CollectionTableSink.CollectionTableSinkFunction.class);

  28.     private static Map<String, List<String>> map = new ConcurrentHashMap<>();

  29.     private DeploySite deploySite;

  30.     private String jobName;

  31.     private static final Object lock = new Object();

  32. 

  33.     public CollectionTableSinkFunction(DeploySite deploySite, String jobName) {

  34.         this.deploySite = deploySite;

  35.         this.jobName = jobName;

  36.     }

  37. 

  38.     @Override

  39.     public void open(Configuration parameters) throws Exception {

  40.         super.open(parameters);

  41.         LOG.info("OPEN!");

  42.     }

  43. 

  44.     @Override

  45.     public void invoke(RowData row, Context context) throws Exception {

  46.         String operatorName = row.getString(0).toString();

  47.         String value = row.getString(1).toString();

  48.         //operatorName source 1

  49. 

  50.         map.compute(operatorName, (k,v) -> {

  51.             if(v==null) {

  52.                 v = new ArrayList<>();

  53.                 v.add(value);

  54.             } else {

  55.                 v.add(value);

  56.             }

  57.             return v;

  58.         });

  59.         LOG.info("put to collectionMap. [{},{}] ", operatorName, value);

  60.         LOG.info("mapSize:" + map.size());

  61.     }

  62. 

  63.     public Map<String,List<String>> getContent() {

  64.         return map;

  65.     }

  66. 

  67.     @Override

  68.     public void close() {

  69.         synchronized (lock) {

  70.             Map<String,List<String>> map = getContent();

  71.             String key = "TalosMockJob_" + jobName;

  72.             String value = JSON.toJSONString(map);

  73.             LOG.info("Send Mock result to redis. key:{}, value:{}", key, value);

  74.             long ttl = 24 * 3600 * 1000;

  75.             try {

  76.                 JedisClient.get(deploySite.name()).setValue(key, value, ttl);

  77.             } catch (Exception e) {

  78. 

  79.             }

  80.         }

  81.     }

  82. }
复制代码
3.最后一步spi


a4.png

把RedisDynamicTableSourceFactory类包路径填进去即可

4.使用
  1.  ctx.getTableEnv().executeSql("create table MockJob_Data_Table (\n" +

  2.                 "    operatorName STRING," +

  3.                 "    data STRING" +

  4.                 ") WITH (\n" +

  5.                 "    'connector' = 'redis'\n" +

  6.                 ")");
复制代码



最新经典文章，欢迎关注公众号



原文链接: https://www.jianshu.com/p/48206a78425d








回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
Mirinda

超级版主

关注

  • 123

    主题

  • 125

    帖子

  • 11

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000