Flink 支持 HiveCatalog 作为表元数据持久化的介质，在生产环境我们一般采用 HiveCatalog 来管理元数据, 这样的好处是不需要重复使用 DDL 创建表,只需要关心业务逻辑的 SQL,简化了开发的流程,可以节省很多时间,下面就来介绍一下怎么配置和使用 HiveCatalog.
sql-client-defaults.yaml 配置
catalogs: - name: myhive type: hive hive-conf-dir: /home/jason/bigdata/hive/hive-2.3.4 default-database: mydatabase
添加依赖
-rw-r--r--. 1 root root 42998 Jul 22 2020 flink-connector-filesystem_2.11-1.11.1.jar -rw-r--r--. 1 root root 196416 Dec 11 17:51 flink-connector-jdbc_2.11-1.12.0.jar -rw-r--r--. 1 root root 91553 Dec 2 17:46 flink-csv-1.12.0.jar -rw-r--r--. 1 root root 114120165 Dec 2 17:50 flink-dist_2.11-1.12.0.jar -rw-r--r--. 1 root root 136663 Dec 2 17:46 flink-json-1.12.0.jar -rw-r--r--. 1 root root 43317025 Dec 11 12:44 flink-shaded-hadoop-2-uber-2.8.3-10.0.jar -rw-r--r--. 1 root root 7709741 Sep 30 01:49 flink-shaded-zookeeper-3.4.14.jar -rw-r--r--. 1 root root 3309441 Dec 12 15:35 flink-sql-avro-1.12.0.jar -rw-r--r--. 1 root root 40650306 Dec 19 12:42 flink-sql-connector-hive-2.3.6_2.11-1.12.0.jar -rw-r--r--. 1 root root 3650212 Dec 11 14:44 flink-sql-connector-kafka_2.11-1.12.0.jar -rw-r--r--. 1 root root 2124047 Dec 12 15:35 flink-sql-orc_2.11-1.12.0.jar -rw-r--r--. 1 root root 5666201 Dec 12 15:35 flink-sql-parquet_2.11-1.12.0.jar -rw-r--r--. 1 root root 36147819 Dec 2 17:49 flink-table_2.11-1.12.0.jar -rw-r--r--. 1 root root 40286358 Dec 2 17:49 flink-table-blink_2.11-1.12.0.jar -rw-r--r--. 1 root root 34214106 Dec 19 19:18 hive-exec-2.3.4.jar -rw-r--r--. 1 root root 67114 Feb 22 2020 log4j-1.2-api-2.12.1.jar -rw-r--r--. 1 root root 276771 Feb 22 2020 log4j-api-2.12.1.jar -rw-r--r--. 1 root root 1674433 Feb 22 2020 log4j-core-2.12.1.jar -rw-r--r--. 1 root root 23518 Feb 22 2020 log4j-slf4j-impl-2.12.1.jar -rw-r--r--. 1 root root 1007502 Dec 11 17:50 mysql-connector-java-5.1.47.jar
代码里面使用 HiveCatalog
package flink.ddl import java.time.ZoneOffset._ import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.table.api.EnvironmentSettings.newInstance import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment.create import org.apache.flink.table.catalog.hive.HiveCatalog import org.apache.flink.table.module.hive.HiveModule import org.apache.hadoop.hive.ql.exec.UDF import org.apache.hadoop.io.Text /** * Flink SQL 使用 hive catalog */ object FlinkDDLHiveCatalog { private val catalog_name = "myhive" private val hive_version = "2.3.4" def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment val settings = newInstance() .useBlinkPlanner() .inStreamingMode() .build() val tEnv = create(env, settings) tEnv.getConfig.setLocalTimeZone(ofHours(8)) // 设置 early fired tEnv.getConfig().getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true) tEnv.getConfig().getConfiguration().setString("table.exec.emit.early-fire.delay", "5000 ms") // 设置 job name tEnv.getConfig.getConfiguration.setString("pipeline.name",this.getClass.getSimpleName.replace("$","")) val catalog = new HiveCatalog( catalog_name, // catalog name "mydatabase", // default database "/home/jason/bigdata/hive/hive-2.3.4", // Hive config (hive-site.xml) directory hive_version // Hive version ) // 注册 catalog tEnv.registerCatalog("myhive", catalog) // 选择一个 catalog tEnv.useCatalog("myhive") // 选择 database tEnv.useDatabase("mydatabase") // 加载 hive 的内置函数 tEnv.loadModule(catalog_name,new HiveModule(hive_version)) // kafka_source_jason 和 print_table 提前已经创建好可以直接使用 tEnv.executeSql( """ |insert into print_table |select |lower(funcName), |MIN(`timestamp`) as min_timestamp, |FIRST_VALUE(`timestamp`) as first_timestamp, |MAX(`timestamp`) as max_timestamp |from kafka_source_jason |group by funcName |""".stripMargin) } }
因为 kafka_source_jason 和 print_table 这两张表提前已经创建过了,已经保存在 HiveCatalog 里面,所以代码里面可以直接使用不用再次创建.
提交任务
在启动任务之前,需要先把 Hiv e的 metastore 起起来,因为 HiveCatalog 会和 metastore 连接这样才能访问元数据信息.
hive --service metastore & flink run -d -m yarn-cluster \ -c flink.ddl.FlinkDDLHiveCatalog \ -yqu flink \ -nm FlinkDDLHiveCatalog \ -p 6 \ /home/jason/bigdata/jar/flink-1.11.1-1.0-SNAPSHOT.jar
打印结果
上面的代码还加载了 hive 的内置函数, Flink SQL 里面可以直接使用 hive 的内置函数, SQL 中的 lower 就是 hive 的函数可以直接拿来使用,这样就会非常的方便.
