Apache Hive

    Flink offers a two-fold integration with Hive.

    The first is to leverage Hive’s Metastore as a persistent catalog with Flink’s for storing Flink specific metadata across sessions. For example, users can store their Kafka or ElasticSearch tables in Hive Metastore by using HiveCatalog, and reuse them later on in SQL queries.

    The second is to offer Flink as an alternative engine for reading and writing Hive tables.

    The HiveCatalog is designed to be “out of the box” compatible with existing Hive installations. You do not need to modify your existing Hive Metastore or change the data placement or partitioning of your tables.

    Flink supports the following Hive versions.

    • 1.0
      • 1.0.0
      • 1.0.1
    • 1.1
      • 1.1.0
      • 1.1.1
    • 1.2
      • 1.2.0
      • 1.2.1
      • 1.2.2
    • 2.0
      • 2.0.0
      • 2.0.1
    • 2.1
      • 2.1.0
      • 2.1.1
    • 2.2
      • 2.2.0
    • 2.3
      • 2.3.0
      • 2.3.1
      • 2.3.2
      • 2.3.3
      • 2.3.4
      • 2.3.5
      • 2.3.6
    • 3.1
      • 3.1.0
      • 3.1.1
      • 3.1.2

    Please note Hive itself have different features available for different versions, and these issues are not caused by Flink:

    • Hive built-in functions are supported in 1.2.0 and later.
    • Column constraints, i.e. PRIMARY KEY and NOT NULL, are supported in 3.1.0 and later.
    • Altering table statistics is supported in 1.2.0 and later.
    • DATE column statistics are supported in 1.2.0 and later.
    • Writing to ORC tables is not supported in 2.0.x.

    To integrate with Hive, you need to add some extra dependencies to the /lib/ directory in Flink distribution to make the integration work in Table API program or SQL in SQL Client. Alternatively, you can put these dependencies in a dedicated folder, and add them to classpath with the -C or -l option for Table API program or SQL Client respectively.

    Apache Hive is built on Hadoop, so you need to provide Hadoop dependencies, by setting the HADOOP_CLASSPATH environment variable:

    There are two ways to add Hive dependencies. First is to use Flink’s bundled Hive jars. You can choose a bundled Hive jar according to the version of the metastore you use. Second is to add each of the required jars separately. The second way can be useful if the Hive version you’re using is not listed here.

    Using bundled hive jar

    The following tables list all available bundled hive jars. You can pick one to the /lib/ directory in Flink distribution.

    User defined dependencies

    Please find the required dependencies for different Hive major versions below.

    Hive 2.3.4

    2. /flink-1.14.4
    3.    /lib
    5.        // Flink's Hive connector.Contains flink-hadoop-compatibility and flink-orc jars
    6.        flink-connector-hive_2.11-1.14.4.jar
    8.        // Hive dependencies
    9.        hive-exec-2.3.4.jar
    11.        // add antlr-runtime if you need to use hive dialect
    12.        antlr-runtime-3.5.2.jar

    Hive 1.0.0

    1. /flink-1.14.4
    2.    /lib
    4.        // Flink's Hive connector
    5.        flink-connector-hive_2.11-1.14.4.jar
    7.        // Hive dependencies
    8.        hive-metastore-1.0.0.jar
    9.        hive-exec-1.0.0.jar
    10.        libfb303-0.9.0.jar // libfb303 is not packed into hive-exec in some versions, need to add it separately
    12.        // Orc dependencies -- required by the ORC vectorized optimizations
    13.        orc-core-1.4.3-nohive.jar
    14.        aircompressor-0.8.jar // transitive dependency of orc-core
    16.        antlr-runtime-3.5.2.jar

    Hive 1.1.0

    1. /flink-1.14.4
    2.    /lib
    4.        // Flink's Hive connector
    5.        flink-connector-hive_2.11-1.14.4.jar
    7.        // Hive dependencies
    8.        hive-metastore-1.1.0.jar
    9.        hive-exec-1.1.0.jar
    10.        libfb303-0.9.2.jar // libfb303 is not packed into hive-exec in some versions, need to add it separately
    12.        // Orc dependencies -- required by the ORC vectorized optimizations
    13.        orc-core-1.4.3-nohive.jar
    14.        aircompressor-0.8.jar // transitive dependency of orc-core
    16.        // add antlr-runtime if you need to use hive dialect
    17.        antlr-runtime-3.5.2.jar

    Hive 1.2.1

    1. /flink-1.14.4
    2.    /lib
    4.        // Flink's Hive connector
    5.        flink-connector-hive_2.11-1.14.4.jar
    7.        // Hive dependencies
    8.        hive-metastore-1.2.1.jar
    9.        hive-exec-1.2.1.jar
    10.        libfb303-0.9.2.jar // libfb303 is not packed into hive-exec in some versions, need to add it separately
    12.        // Orc dependencies -- required by the ORC vectorized optimizations
    13.        orc-core-1.4.3-nohive.jar
    14.        aircompressor-0.8.jar // transitive dependency of orc-core
    16.        // add antlr-runtime if you need to use hive dialect
    17.        antlr-runtime-3.5.2.jar

    Hive 2.0.0

    Hive 2.1.0

    1. /flink-1.14.4
    2.    /lib
    4.        // Flink's Hive connector
    5.        flink-connector-hive_2.11-1.14.4.jar
    7.        // Hive dependencies
    8.        hive-exec-2.1.0.jar
    10.        // add antlr-runtime if you need to use hive dialect
    11.        antlr-runtime-3.5.2.jar

    Hive 2.2.0

    1. /flink-1.14.4
    2.    /lib
    4.        // Flink's Hive connector
    5.        flink-connector-hive_2.11-1.14.4.jar
    7.        // Hive dependencies
    8.        hive-exec-2.2.0.jar
    10.        // Orc dependencies -- required by the ORC vectorized optimizations
    11.        orc-core-1.4.3.jar
    12.        aircompressor-0.8.jar // transitive dependency of orc-core
    14.        // add antlr-runtime if you need to use hive dialect
    15.        antlr-runtime-3.5.2.jar
    1. /flink-1.14.4
    2.    /lib
    4.        // Flink's Hive connector
    5.        flink-connector-hive_2.11-1.14.4.jar
    7.        // Hive dependencies
    8.        hive-exec-3.1.0.jar
    9.        libfb303-0.9.3.jar // libfb303 is not packed into hive-exec in some versions, need to add it separately
    10.        // add antlr-runtime if you need to use hive dialect
    11.        antlr-runtime-3.5.2.jar

    Program maven

    If you are building your own program, you need the following dependencies in your mvn file. It’s recommended not to include these dependencies in the resulting jar file. You’re supposed to add dependencies as stated above at runtime.

    1. <!-- Flink Dependency -->
    2. <dependency>
    3.   <groupId>org.apache.flink</groupId>
    4.   <artifactId>flink-connector-hive_2.11</artifactId>
    5.   <version>1.14.4</version>
    6.   <scope>provided</scope>
    7. </dependency>
    10.   <groupId>org.apache.flink</groupId>
    11.   <artifactId>flink-table-api-java-bridge_2.11</artifactId>
    12.   <version>1.14.4</version>
    13.   <scope>provided</scope>
    14. </dependency>
    16. <!-- Hive Dependency -->
    17. <dependency>
    18.     <groupId>org.apache.hive</groupId>
    19.     <artifactId>hive-exec</artifactId>
    20.     <version>${hive.version}</version>
    21.     <scope>provided</scope>
    22. </dependency>

    Connect to an existing Hive installation using the and HiveCatalog through the table environment or YAML configuration.

    Following is an example of how to connect to Hive:

    Java

    Scala

    2. val settings = EnvironmentSettings.inStreamingMode()
    3. val tableEnv = TableEnvironment.create(settings)
    5. val name            = "myhive"
    6. val defaultDatabase = "mydatabase"
    7. val hiveConfDir     = "/opt/hive-conf"
    9. val hive = new HiveCatalog(name, defaultDatabase, hiveConfDir)
    10. tableEnv.registerCatalog("myhive", hive)
    12. // set the HiveCatalog as the current catalog of the session
    13. tableEnv.useCatalog("myhive")

    Python

    1. from pyflink.table import *
    2. from pyflink.table.catalog import HiveCatalog
    4. settings = EnvironmentSettings.in_batch_mode()
    5. t_env = TableEnvironment.create(settings)
    7. catalog_name = "myhive"
    8. default_database = "mydatabase"
    9. hive_conf_dir = "/opt/hive-conf"
    11. hive_catalog = HiveCatalog(catalog_name, default_database, hive_conf_dir)
    12. t_env.register_catalog("myhive", hive_catalog)
    14. # set the HiveCatalog as the current catalog of the session
    15. tableEnv.use_catalog("myhive")

    YAML

    2. execution:
    3.     ...
    4.     current-catalog: myhive  # set the HiveCatalog as the current catalog of the session
    5.     current-database: mydatabase
    7. catalogs:
    8.    - name: myhive
    9.      type: hive
    10.      hive-conf-dir: /opt/hive-conf

    SQL

    2. CREATE CATALOG myhive WITH (
    3.     'type' = 'hive',
    4.     'default-database' = 'mydatabase',
    5.     'hive-conf-dir' = '/opt/hive-conf'
    6. );
    7. -- set the HiveCatalog as the current catalog of the session
    8. USE CATALOG myhive;

    Below are the options supported when creating a HiveCatalog instance with YAML file or DDL.

    OptionRequiredDefaultTypeDescription
    type
    		Yes(none)StringType of the catalog. Must be set to when creating a HiveCatalog.
    name
    		Yes(none)StringThe unique name of the catalog. Only applicable to YAML file.
    hive-conf-dir
    		No(none)StringURI to your Hive conf dir containing hive-site.xml. The URI needs to be supported by Hadoop FileSystem. If the URI is relative, i.e. without a scheme, local file system is assumed. If the option is not specified, hive-site.xml is searched in class path.
    default-database
    		NodefaultStringThe default database to use when the catalog is set as the current catalog.
    hive-version
    		No(none)StringHiveCatalog is capable of automatically detecting the Hive version in use. It’s recommended NOT to specify the Hive version, unless the automatic detection fails.
    hadoop-conf-dir
    		No(none)StringPath to Hadoop conf dir. Only local file system paths are supported. The recommended way to set Hadoop conf is via the HADOOP_CONF_DIR environment variable. Use the option only if environment variable doesn’t work for you, e.g. if you want to configure each HiveCatalog separately.

    Flink supports DML writing to Hive tables. Please refer to details in