SparkSql使用Catalog连接Hive

阳光穿透心脏的1/2处 2022-11-07 14:53 256阅读 0赞

# SparkSql使用Catalog连接Hive #

### 目录 ###

*  SparkSql使用Catalog连接Hive
 *   *  一、Catalog
     *  二、Catalog类型
     *  三、HiveCatalog连接Hive获取数据

## 一、Catalog ##

Catalog 提供了`元数据信息`，例如数据库、表、分区、视图以及数据库或其他外部系统中存储的函数和信息。

数据处理最关键的方面之一是`管理元数据`。 元数据可以是临时的，例如临时表、或者通过 TableEnvironment 注册的 UDF。 元数据也可以是持久化的，例如 Hive Metastore 中的元数据。Catalog 提供了一个统一的API，用于管理元数据，并使其可以从 Table API 和 SQL 查询语句中来访问。

## 二、Catalog类型 ##

*  GenericInMemoryCatalog  
    GenericInMemoryCatalog 是基于内存实现的 Catalog，所有元数据只在 session 的生命周期内可用。
 *  JdbcCatalog  
    JdbcCatalog 使得用户可以将 Flink 通过 JDBC 协议连接到关系数据库。PostgresCatalog 是当前实现的唯一一种 JDBC Catalog。
 *  HiveCatalog  
    HiveCatalog 有两个用途：作为原生 Flink 元数据的持久化存储，以及作为读写现有 Hive 元数据的接口。 Flink 的 Hive 文档 提供了有关设置 HiveCatalog 以及访问现有 Hive 元数据的详细信息。

## 三、HiveCatalog连接Hive获取数据 ##

**导入需要的依赖**

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-exec</artifactId>
        <version>3.1.2</version>
    </dependency>

**在hadoop162启动hive元数据**

nohup hive --service metastore >/dev/null 2>&1 &

**连接 Hive**

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
    String name            = "myhive";  // Catalog 名字
    String defaultDatabase = "flink_test"; // 默认数据库
    String hiveConfDir     = "c:/conf"; // hive配置文件的目录. 需要把hive-site.xml添加到该目录
    
    // 1. 创建HiveCatalog
    HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
    // 2. 注册HiveCatalog
    tEnv.registerCatalog(name, hive);
    // 3. 把 HiveCatalog: myhive 作为当前session的catalog
    tEnv.useCatalog(name);
    tEnv.useDatabase("flink_test");
    tEnv.sqlQuery("select * from stu").execute().print();