Flink DataStream Join小规模维度数据的简便方法

2019/11/26 19:22
阅读数 11

在编写基于Flink的ETL程序时,我们经常需要用维度数据丰富我们接入的流式数据,如通过商品ID获得商品名称、通过商品分类ID获得分类名称等等。而维度表基本都位于外部存储,换句话说,就是要解决一个无界的流式表与一个有界的码表或半静态表做join操作的问题。

一般情况下的首选方案是Flink内置的异步I/O机制,必要时还得配合使用高效的缓存(如Guava提供的LoadingCache)减少对外部数据源的请求压力。由于今天时间紧张,所以不深入谈它的原理和用法了,之后会再提。看官如果想了解的话,可以先参考官方文档和FLIP-12给出的设计细节。

但是,异步I/O对于那种变化缓慢并且规模不大的维度数据,就显得有些杀鸡用牛刀了。我们完全可以自己做个轻量级的实现。下面举出一个示例,它从订单日志中取出站点ID、城市ID,然后从存储在MySQL的维度表中获取站点名和城市名,并写回订单日志。

  public static final class MapWithSiteInfoFunc
    extends RichMapFunction<String, String> {
    private static final Logger LOGGER = LoggerFactory.getLogger(MapWithSiteInfoFunc.class);
    private static final long serialVersionUID = 1L;

    private transient ScheduledExecutorService dbScheduler;
    private Map<Integer, SiteAndCityInfo> siteInfoCache;

    @Override
    public void open(Configuration parameters) throws Exception {
      super.open(parameters);
      siteInfoCache = new HashMap<>(1024);

      dbScheduler = new ScheduledThreadPoolExecutor(1, r -> {
        Thread thread = new Thread(r, "site-info-update-thread");
        thread.setUncaughtExceptionHandler((t, e) -> {
          LOGGER.error("Thread " + t + " got uncaught exception: " + e);
        });
        return thread;
      });

      dbScheduler.scheduleWithFixedDelay(() -> {
        try {
          QueryRunner queryRunner = new QueryRunner(JdbcUtil.getDataSource());
          List<Map<String, Object>> info = queryRunner.query(SITE_INFO_QUERY_SQL, new MapListHandler());

          for (Map<String, Object> item : info) {
            siteInfoCache.put((int) item.get("site_id"), new SiteAndCityInfo(
              (int) item.get("site_id"),
              (String) item.getOrDefault("site_name", ""),
              (long) item.get("city_id"),
              (String) item.getOrDefault("city_name", "")
            ));
          }

          LOGGER.info("Fetched {} site info records, {} records in cache", info.size(), siteInfoCache.size());
        } catch (Exception e) {
          LOGGER.error("Exception occurred when querying: " + e);
        }
      }, 0, 10 * 60, TimeUnit.SECONDS);
    }

    @Override
    public String map(String value) throws Exception {
      JSONObject json = JSON.parseObject(value);
      int siteId = json.getInteger("site_id");
     
      String siteName = "", cityName = "";
      SiteAndCityInfo info = siteInfoCache.getOrDefault(siteId, null);
      if (info != null) {
        siteName = info.getSiteName();
        cityName = info.getCityName();
      }

      json.put("site_name", siteName);
      json.put("city_name", cityName);
      return json.toJSONString();
    }

    @Override
    public void close() throws Exception {
      siteInfoCache.clear();
      ExecutorUtils.gracefulShutdown(10, TimeUnit.SECONDS, dbScheduler);
      JdbcUtil.close();

      super.close();
    }

    private static final String SITE_INFO_QUERY_SQL = "...";
  }

这段代码的思路很直接:用一个RichMapFunction封装整个join过程,用一个单线程的调度线程池每隔10分钟请求MySQL,拉取想要的维度表数据存入HashMap,再根据日志中的ID查HashMap就完事了。为了安全,在RichMapFunction的close()方法里要记得关闭线程池和连接。

上述代码中的QueryRunner和MapListHandler来自Apache Commons框架里的JDBC工具DBUtils。JdbcUtil中则封装了MySQL连接的参数与DBCP2里的基本连接池BasicDataSource,很简单,看官可以自行实现。

声明:本号所有文章除特殊注明,都为原创,公众号读者拥有优先阅读权,未经作者本人允许不得转载,否则追究侵权责任。

关注我的公众号,后台回复【JAVAPDF】获取200页面试题! 5万人关注的大数据成神之路,不来了解一下吗? 5万人关注的大数据成神之路,真的不来了解一下吗? 5万人关注的大数据成神之路,确定真的不来了解一下吗?

欢迎您关注《大数据成神之路》

大数据技术与架构

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部