Flink DataStream Join小規模維度數據的簡便方法

在編寫基於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萬人關注的大數據成神之路，確定真的不來了解一下嗎？

歡迎您關注《大數據成神之路》