devbox
Cpp五条
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink 实战 —— Flink 集成 Hive_flink集成hive

作者：Cpp五条 | 2024-04-29 19:11:12

赞

踩

flink集成hive

文章目录


Hive 是大数据领域最早出现的 SQL 引擎，发展至今有着丰富的功能和广泛的用户基础。之后出现的 SQL 引擎，如 Spark SQL、Impala 等，都在一定程度上提供了与 Hive 集成的功能，从而方便用户使用现有的数据仓库、进行作业迁移等。

Flink从1.9开始支持集成Hive，不过1.9版本为beta版，不推荐在生产环境中使用。在最新版Flink1.10版本，标志着对 Blink的整合宣告完成，达到了对 Hive 的生产级别集成，Hive作为数据仓库系统的绝对核心，承担着绝大多数的离线数据ETL计算和数据管理，期待Flink未来对Hive的完美支持。

而 HiveCatalog 会与一个 Hive Metastore 的实例连接，提供元数据持久化的能力。要使用 Flink 与 Hive 进行交互，用户需要配置一个 HiveCatalog，并通过 HiveCatalog 访问 Hive 中的元数据。

1. 程序集成 Hive

1.1 添加依赖

要与Hive集成，需要在Flink的lib目录下添加额外的依赖jar包，以使集成在Table API程序或SQL Client中的SQL中起作用。或者，可以将这些依赖项放在文件夹中，并分别使用Table API程序或SQL Client 的-C 或-l选项将它们添加到classpath中。本文使用第一种方式，即将jar包直接复制到$FLINK_HOME/lib目录下。本文使用的Hive版本为1.2.1(对于不同版本的Hive，可以参照官网选择不同的jar包依赖)，总共需要以下个jar包，如下：

  • antlr-runtime-3.4.jar
  • datanucleus-api-jdo-3.2.6.jar
  • datanucleus-core-3.2.10.jar
  • datanucleus-rdbms-3.2.9.jar
  • flink-connector-hive_2.12-1.10.2.jar
  • flink-dist_2.12-1.10.2.jar
  • flink-shaded-hadoop-2-uber-2.6.5-8.0.jar
  • flink-table_2.12-1.10.2.jar
  • flink-table-blink_2.12-1.10.2.jar
  • flink-table-planner_2.12-1.10.2.jar
  • flink-table-planner-blink_2.12-1.10.2.jar
  • hive-exec-1.2.1.jar
  • hive-metastore-1.2.1.jar
  • jdo-api-3.0.1.jar
  • libfb303-0.9.2.jar
  • log4j-1.2.17.jar
  • mysql-connector-java-5.1.44.jar
  • slf4j-log4j12-1.7.15.jar

1.2 构建程序

  1. 添加 pom.xml依赖

    <!-- Flink Dependency -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-hive_2.11</artifactId>
  <version>1.10.2</version>
  <scope>provided</scope>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-java-bridge_2.11</artifactId>
  <version>1.10.2</version>
  <scope>provided</scope>
</dependency>

<!-- Hive Dependency -->
<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-exec</artifactId>
    <version>${hive.version}</version>
    <scope>provided</scope>
</dependency>  

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

  2. 编写程序

    package com.flink.hive;

import lombok.val;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.scala.DataSet;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.api.scala.TableConversions;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.module.Module;
import org.apache.flink.table.module.hive.HiveModule;
import org.apache.flink.table.sinks.CsvTableSink;
import org.apache.flink.table.sinks.TableSink;
import org.apache.flink.types.Row;
import javax.jdo.JDOException;
public class FlinkHive {

    public static void main(String[] args) throws Exception{
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .useBlinkPlanner() // 使用BlinkPlanner
                .inBatchMode() // Batch模式，默认为StreamingMode
                .build();

        //使用StreamingMode
       /* EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .useBlinkPlanner() // 使用BlinkPlanner
                .inStreamingMode() // StreamingMode
                .build();*/
         TableEnvironment tableEnv = TableEnvironment.create(settings);
        // 定义一个 sink
        CsvTableSink tableSink = new CsvTableSink("/opt/module/flink/ads_uv_count.log");
        TypeInformation<?>[] fieldTypes = {
                TypeInformation.of(String.class),
                TypeInformation.of(Long.class),
                TypeInformation.of(Long.class),
                TypeInformation.of(Long.class),
                TypeInformation.of(String.class),
                TypeInformation.of(String.class)
        };

        // 定义 sink 的字段
        String[] fieldNames = {"dt","day_count","wk_count","mn_count","is_weekend","is_monthend"};
        TableSink<Row> configure = tableSink.configure(fieldNames, fieldTypes);
        
        // Catalog名称，定义一个唯一的名称表示
        String cataLogName = "ads_uv_count";
        // 默认数据库
        String defaultDatabase = "gmall";
        // hive-site.xml路径
        String hiveConfDir = "/opt/module/hive/conf/";
        // hive 版本号
        String version = "1.2.1";

        HiveCatalog hive = new HiveCatalog(cataLogName, defaultDatabase, hiveConfDir, version);
        tableEnv.registerCatalog(cataLogName, hive);
        tableEnv.useCatalog(cataLogName);
        String sql = "select dt,day_count,wk_count,mn_count,is_weekend,is_monthend from ads_uv_count";
        Table table = tableEnv.sqlQuery(sql);
        // 注册 sink 表
        tableEnv.registerTableSink("ads_uv_count_sink", configure);

        // 把数据写入 sink 表
        table.insertInto("ads_uv_count_sink");

        tableEnv.execute("操作hive");

    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78

2. Flink SQL Client集成Hive

Flin k的表和 SQL API 可以处理用 SQL 语言编写的查询，但是这些查询需要嵌入到用 Java 或 Scala 编写的程序中。此外，这些程序在提交到集群之前需要与构建工具打包。这或多或少地限制了 Java/Scala 程序员对 Flink 的使用。

SQL 客户端旨在提供一种简单的方式，无需一行 Java 或 Scala 代码，即可将表程序编写、调试和提交到 Flink 集群。Flink SQL 客户端 CLI 允许通过命令行的形式运行分布式程序。使用 Flink SQL cli 访问 Hive，需要配置 sql-client-defaults.yaml文件或指定默认配置文件。

2.1 sql-client-defaults.yaml 配置

目前 Hive TableSink 不支持流式写入（未实现 AppendStreamTableSink）。需要将执行模式改成 batch
模式，否则会报如下错误：

org.apache.flink.table.api.TableException: Stream Tables can only be emitted by AppendStreamTableSink, RetractStream

  • 1

需要修改的配置内容如下：

#...省略的配置项...

#==============================================================================
# Catalogs
#==============================================================================
# 配置catalogs,可以配置多个.
catalogs: # empty list
  - name: gmall
    type: hive
    hive-conf-dir: /opt/module/hive/conf/
    hive-version: 1.2.1
    default-database: gmall

#...省略的配置项...

#==============================================================================
# Execution properties
#==============================================================================

# Properties that change the fundamental execution behavior of a table program.

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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27

2.2 启动 Flink SQL Cli

启动默认配置

$ bin/sql-client.sh embedded

  • 1

启动指定配置

$ bin/sql-client.sh embedded -d conf/sql-client-hive.yaml

  • 1

在启动之前，确保Hive的 metastore 已经开启了，否则会报 Failed to create Hive Metastore client异常。启动成功，如下图：
在这里插入图片描述

启动之后，就可以在此Cli下执行SQL命令访问Hive的表了，基本的操作如下：

-- 命令行帮助
Flink SQL> help
-- 查看当前会话的catalog，其中myhive为自己配置的，default_catalog为默认的
Flink SQL> show catalogs;
default_catalog
gmall
-- 使用catalog
Flink SQL> use catalog gmall;
-- 查看当前catalog的数据库
Flink SQL> show databases;
-- 创建数据库
Flink SQL> create database testdb;
-- 删除数据库
Flink SQL> drop database testdb;
-- 创建表
Flink SQL> create table tbl(id int,name string);
-- 删除表
Flink SQL> drop table tbl;
-- 查询表
Flink SQL> select * from  code_city;
-- 插入数据
Flink SQL> insert overwrite code_city select id,city,province,event_time from code_city_delta ;
Flink SQL> INSERT into code_city values(1,'南京','江苏','');

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

3. 小结

本文以最新版本的 Flink 为例，对 Flink 集成 Hive 进行了实操。首先通过代码的方式与 Hive 进行集成，然后介绍了如何使用 Flink SQL 客户端访问 Hive，并对其中会遇到的坑进行了描述，最后给出了Flink SQL Cli 的详细使用。相信在未来的版本中 Flink SQL 会越来越完善，期待F link 未来对 Hive 的完美支持。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Cpp五条/article/detail/509321
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号