zoukankan      html  css  js  c++  java
  • Flink如何做维表关联?

    使用

    RichAsyncFunction 加 CacheBuilder

    
    CacheBuilder.newBuilder()
            //最多存储10000条
            .maximumSize(10000)
            //过期时间为1分钟
            .expireAfterWrite(60, TimeUnit.SECONDS)
            .build();
    
    public class LRU extends RichAsyncFunction<String,Order> {
        private static final Logger LOGGER = LoggerFactory.getLogger(LRU.class);
        String table = "info";
        Cache<String, String> cache = null;
        private HBaseClient client = null;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            //创建hbase客户端
            client = new HBaseClient("127.0.0.1","7071");
            cache = CacheBuilder.newBuilder()
                    //最多存储10000条
                    .maximumSize(10000)
                    //过期时间为1分钟
                    .expireAfterWrite(60, TimeUnit.SECONDS)
                    .build();
        }
        @Override
        public void asyncInvoke(String input, ResultFuture<Order> resultFuture) throws Exception {
            JSONObject jsonObject = JSONObject.parseObject(input);
            Integer cityId = jsonObject.getInteger("city_id");
            String userName = jsonObject.getString("user_name");
            String items = jsonObject.getString("items");
            //读缓存
            String cacheCityName = cache.getIfPresent(cityId);
            //如果缓存获取失败再从hbase获取维度数据
            if(cacheCityName != null){
                Order order = new Order();
                order.setCityId(cityId);
                order.setItems(items);
                order.setUserName(userName);
                order.setCityName(cacheCityName);
                resultFuture.complete(Collections.singleton(order));
            }else {
                client.get(new GetRequest(table,String.valueOf(cityId))).addCallback((Callback<String, ArrayList<KeyValue>>) arg -> {
                    for (KeyValue kv : arg) {
                        String value = new String(kv.value());
                        Order order = new Order();
                        order.setCityId(cityId);
                        order.setItems(items);
                        order.setUserName(userName);
                        order.setCityName(value);
                        resultFuture.complete(Collections.singleton(order));
                        cache.put(String.valueOf(cityId), value);
                    }
                    return null;
                });
            }
        }
    }
    
    
  • 相关阅读:
    Mysql Explain 详解【转】
    Windows下Gradle安装与配置
    MYSQ创建联合索引,字段的先后顺序,对查询的影响分析
    给.Net 5 Api增加JwtBearer认证
    纸壳CMS 3.3.6发布升级.Net 5
    PL/SQL Developer登入时候报ORA-12638: 身份证明检索失败的解决办法
    Mybatis 日志工厂
    Mybatis 配置解析
    Mybatis 完成增删改查
    Mybatis 简介
  • 原文地址:https://www.cnblogs.com/weijiqian/p/14034479.html
Copyright © 2011-2022 走看看