京东|京东：Flink SQL 优化实战( 二 )

大家可能会好奇为什么会把单元测试也放到优化里面，大家都知道 Flink 任务调试过程非常复杂，对于一些线上机器隔离的公司来说尤甚。京东的本地环境是没有办法访问任务服务器的，因此在初始阶段调试任务，我们耗费了很多时间用来上传 jar 包、查看日志等行为。
为了降低任务的调试时间、增加代码开发人员的开发效率，实现了内嵌式的 Derby 来作为 Hive 的元数据存储数据库 (allowEmbedded) ，这算是一种优化开发时间的方法。具体思路如下：
首先创建 Hive Conf：
public static HiveConf createHiveConf() { ClassLoader classLoader = new HiveOperatorTest().getClass().getClassLoader(); HiveConf.setHiveSiteLocation(classLoader.getResource(HIVE_SITE_XML)); try { TEMPORARY_FOLDER.create(); String warehouseDir = TEMPORARY_FOLDER.newFolder().getAbsolutePath() + \"/metastore_db\"; String warehouseUri = String.format(HIVE_WAREHOUSE_URI_FORMAT warehouseDir); HiveConf hiveConf = new HiveConf(); hiveConf.setVar( HiveConf.ConfVars.METASTOREWAREHOUSE TEMPORARY_FOLDER.newFolder(\"hive_warehouse\").getAbsolutePath()); hiveConf.setVar(HiveConf.ConfVars.METASTORECONNECTURLKEY warehouseUri); hiveConf.set(\"datanucleus.connectionPoolingType\" \"None\"); hiveConf.set(\"hive.metastore.schema.verification\" \"false\"); hiveConf.set(\"datanucleus.schema.autoCreateTables\" \"true\"); return hiveConf;catch (IOException e) { throw new CatalogException(\"Failed to create test HiveConf to HiveCatalog.\" e);接下来创建 Hive Catalog：(利用反射的方式调用 embedded 的接口)
public static void createCatalog() throws Exception{ Class clazz = HiveCatalog.class; Constructor c1 = clazz.getDeclaredConstructor(new Class[
{String.class String.class HiveConf.class String.class boolean.class); c1.setAccessible(true); hiveCatalog = (HiveCatalog)c1.newInstance(new Object[
{\"test-catalog\" null createHiveConf() \"2.3.4\" true); hiveCatalog.open(); 创建 tableEnvironment：(同官网)
EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();TableEnvironment tableEnv = TableEnvironment.create(settings);TableConfig tableConfig = tableEnv.getConfig();Configuration configuration = new Configuration();configuration.setInteger(\"table.exec.resource.default-parallelism\" 1);tableEnv.registerCatalog(hiveCatalog.getName() hiveCatalog);tableEnv.useCatalog(hiveCatalog.getName()); 最后关闭 Hive Catalog:
public static void closeCatalog() { if (hiveCatalog != null) { hiveCatalog.close();此外，对于单元测试，构建合适的数据集也是一个非常大的功能，我们实现了 CollectionTableFactory ，允许自己构建合适的数据集，使用方法如下：
CollectionTableFactory.reset();CollectionTableFactory.initData(Arrays.asList(Row.of(\"this is a test\") Row.of(\"zhangying480\") Row.of(\"just for test\") Row.of(\"a test case\")));StringBuilder sbFilesSource = new StringBuilder();sbFilesSource.append(\"CREATE temporary TABLE db1.`search_realtime_table_dump_p13`(\" + \" `pvid` string) with ('connector.type'='COLLECTION''is-bounded' = 'true')\");tableEnv.executeSql(sbFilesSource.toString()); 3. join 方式的选择
传统的离线 Batch SQL (面向有界数据集的 SQL) 有三种基础的实现方式，分别是 Nested-loop Join、Sort-Merge Join 和 Hash Join 。

效率
空间
备注
Nested-loop Join
差
占用大

Sort-Merge Join
有sort merge开销
占用小
有序数据集的一种优化措施
Hash Join
高
占用大
适合大小表
Nested-loop Join 最为简单直接，将两个数据集加载到内存，并用内嵌遍历的方式来逐个比较两个数据集内的元素是否符合 Join 条件。 Nested-loop Join 的时间效率以及空间效率都是最低的，可以使用：table.exec.disabled-operators:NestedLoopJoin 来禁用。以下两张图片是禁用前和禁用后的效果 (如果你的禁用没有生效，先看一下是不是 Equi-Join)： Sort-Merge Join 分为 Sort 和 Merge 两个阶段：首先将两个数据集进行分别排序，然后再对两个有序数据集分别进行遍历和匹配，类似于归并排序的合并。 (Sort-Merge Join 要求对两个数据集进行排序，但是如果两个输入是有序的数据集，则可以作为一种优化方案) 。Hash Join 同样分为两个阶段：首先将一个数据集转换为 Hash Table ，然后遍历另外一个数据集元素并与 Hash Table 内的元素进行匹配。第一阶段和第一个数据集分别称为 build 阶段和 build table；第二个阶段和第二个数据集分别称为 probe 阶段和 probe table 。 Hash Join 效率较高但是对空间要求较大，通常是作为 Join 其中一个表为适合放入内存的小表的情况下的优化方案 (并不是不允许溢写磁盘) 。注意：Sort-Merge Join 和 Hash Join 只适用于 Equi-Join ( Join 条件均使用等于作为比较算子) 。