如何在Flink 1.9中使用 Hive？

类似的，也可以通过 Table API 来读写上面提到的这张表。下面的代码展示了如何实现这一操作。

TableEnvironment tableEnv = …; // create TableEnvironment 
tableEnv.registerCatalog("myhive", hiveCatalog); 
// set myhive as current catalog 
tableEnv.useCatalog("myhive"); 
 
Table src = tableEnv.sqlQuery("select * from src"); 
// write src into a sink or do further analysis 
…… 
 
tableEnv.sqlUpdate("insert into src values ('newKey', 'newVal')"); 
tableEnv.execute("insert into src"); 

支持不同的 Hive 版本

Flink 1.9.0 中支持的 Hive 版本是 2.3.4 和 1.2.1，目前我们只针对这两个版本进行了测试。使用 SQL Client 时，如果用户没有在 sql-client-defaults.yaml 文件中指定 Hive 版本，我们会自动检测 classpath 中的 Hive 版本。如果检测到的 Hive 版本不是 2.3.4 或 1.2.1 就会报错。

借助 Hive 兼容性的保证，其它不同的小版本也比较可能是可以正常工作的。因此，如果用户使用的 Hive 小版本与我们所支持的不同，可以指定一个支持的版本来试用与 Hive 集成的功能。比如用户使用的 Hive 版本是 2.3.3，可以在 sql-client-defaults.yaml 文件或者代码中将 Hive 版本指定为 2.3.4。

执行模式与 Planner 的选择

Flink 1.9.0 中 Hive 的 TableSink 只能在 batch 模式下工作，因此如果用户想要使用 Hive 的 TableSink，需要将执行模式设置为 batch。

Flink 1.9.0 增加了新的 blink planner，由于 blink planner 相比于原来的 planner 功能更加全面，因此我们建议在使用 FlinkSQL 与 Hive 集成时使用 blink planner。后续新的功能也可能会只支持 blink planner。

使用 SQL Client 时可以像这样在 sql-client-defaults.yaml 中指定执行模式和 planner：

execution: 
  # select the implementation responsible for planning table programs 
  # possible values are 'old' (used by default) or 'blink' 
  planner: blink 
  # 'batch' or 'streaming' execution 
  type: batch 

对应的 Table API 的写法如下：

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build(); 
TableEnvironment tableEnv = TableEnvironment.create(settings); 

后期规划

我们会在 Flink 后续版本中进一步完善与 Hive 集成的功能，预计会在 1.10.0 版本中实现 Production-Ready。我们在后续版本中计划开展的工作包括：

• 更完整的数据类型支持
• 支持写分区表，包括静态和动态分区
• 支持 INSERT OVERWRITE
• 支持 View
• 更完整的 DDL、DML 的支持
• 支持 Hive 的 TableSink 在 streaming 模式下工作，以便用户将流式数据写入到 Hive 中
• 测试并支持更多的 Hive 版本
• 支持 Bucket 表
• 性能测试与优化

（编辑：核心网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!