Docs

Streaming Computing Flink Version

Catalog Reference

Hive Catalog

The Hive connector provides read and write access to Hive data sources. Flink can use a Hive Catalog to perform unified batch and stream processing on Hive tables. This means Flink can serve as a high-performance alternative to the Hive batch engine or stream data into Hive tables for real-time Data Warehouse applications.

Limitations

The Hive connector is only supported on engine version Flink 1.16-volcano or later.
Streaming to Hive does not support insert overwrite.

Creating Tables

In Flink SQL syntax, data tables are represented by a three-part identifier: Catalog.Database.Table. For a Hive data source, the catalog is a Hive Catalog.
When streaming to Hive tables, you need to set table parameters. See Parameters for stream write at table level.

Task Dependencies

Dependency Files

Hive Catalog requires the following dependency files:

hive-site.xml
hdfs-site.xml

Note: For E-MapReduce clusters, you can obtain these two files from the $HIVE_CONF_DIR directory on the master node.
The upload result is as follows:

Custom Parameters

To resolve the EMR-Hive domain name, set the following two custom parameters:

containerized.master.env.ENV_SEARCH_DOMAIN:emr-volces.com
containerized.taskmanager.env.ENV_SEARCH_DOMAIN:emr-volces.com

The configuration result is as follows:

Parameters

Parameters for creating a catalog

Parameter	Required	Default Value	Type	Description
type	Yes	(None)	String	The type of the catalog. This parameter must be set to `'hive'` when creating a HiveCatalog.
name	Yes	(None)	String	The name of the Catalog. Required only when using a YAML file.
hive-conf-dir	No	(None)	String	A URI pointing to the directory that contains hive-site.xml. The URI must be of a type supported by the Hadoop file system. A relative URI without a scheme defaults to the local file system. If this parameter is not specified, the system searches for hive-site.xml in the class path. tip Enter `/opt/tiger/workdir` if the file is uploaded to a Flink job as a dependency.
default-database	No	default	String	The default database to use when a catalog is set as the current one.
hive-version	No	None	String	HiveCatalog automatically detects the Hive version in use. We recommend not setting the Hive version manually unless automatic detection fails.
hadoop-conf-dir	No	(None)	String	The path to the Hadoop configuration directory. Only local file system paths are currently supported. We recommend using the HADOOP_CONF_DIR environment variable to specify the Hadoop configuration. Therefore, use this parameter only when the environment variable is insufficient, such as when you need to set separate Hadoop configurations for each HiveCatalog.

Job-level parameters for batch reading and writing

Parameter	Required	Default value	Data type	Description
table.exec.hive.fallback-mapred-reader	No	true	Boolean	Specifies whether to enable vectorized reading. Flink automatically performs vectorized reading on Hive tables when the following conditions are met: Format: ORC or Parquet. There are no columns with complex types, such as the Hive column types: List, Map, Struct, or Union. This feature is enabled by default. To disable it, set it to false.
table.exec.hive.infer-source-parallelism	No	true	Boolean	Sets whether to enable source parallelism inference. By default, Flink infers the optimal parallelism for reading Hive based on the number of files and the number of blocks in each file. Flink allows for flexible configuration of the parallelism inference strategy. If this parameter is true, the source parallelism is inferred based on the number of splits. If false, the source parallelism is determined by the configuration. tip This parameter affects all Hive sources in the current job.
table.exec.hive.infer-source-parallelism.max	No	1000	Integer	Sets the maximum inferred parallelism for the source operator. tip This parameter affects all Hive sources in the current job.
table.exec.hive.split-max-size	No	128mb	MemorySize	Sets the data split size when reading Hive tables. When reading Hive tables, data files are divided into multiple splits. Each split is a portion of the data to be read. Splits are the basic unit that Flink uses for task assignment and parallel data reading. You can use this parameter to adjust the size of each split to tune read performance. This parameter specifies the maximum size in bytes for each split when reading Hive tables (default: 128 MB). tip Applies only to Hive tables in ORC format.
table.exec.hive.file-open-cost	No	4mb	MemorySize	The estimated cost in bytes of opening a file when reading from a Hive table. The default is 4 MB. A larger value causes Flink to create fewer splits, which is useful when a Hive table contains many small files. Conversely, a smaller value leads to more splits, which improves data reading parallelism. tip Applies only to Hive tables in ORC format.
table.exec.hive.calculate-partition-size.thread-num	No	3	Integer	The number of threads used to calculate the size of all files in all partitions. To adjust data split sizes, Flink first calculates the total size of files across all partitions. This process can be time-consuming when there are many partitions. Increase the value of this parameter to use more threads and speed up the process.
table.exec.hive.load-partition-splits.thread-num	No	3	Integer	The number of threads for generating splits. Flink uses multiple threads to concurrently split Hive partitions into multiple splits for reading. Use this parameter to configure the number of threads to accelerate the split generation process.
table.exec.hive.sink.statistic-auto-gather.enable	No	true	Boolean	Specifies whether to enable automatic statistics collection. By default, Flink automatically collects statistics for data written to Hive tables and commits them to the Hive metastore. However, you might not want to automatically collect statistics in some cases, as the process can be time-consuming. Set this parameter to false to prevent Flink from automatically gathering statistics. tip Automatic statistics collection is only supported in batch write mode. It is not currently supported in stream write mode.
table.exec.hive.sink.statistic-auto-gather.thread-num	No	3	Integer	Sets the number of threads for gathering statistics. For tables in Parquet or ORC format, Flink only reads the file footers to quickly gather numRows and rawDataSize statistics. However, this process can be time-consuming when there are many files. You can set this parameter to a larger value to accelerate statistics gathering. tip Automatic statistics collection is only supported in batch write mode. It is not currently supported in stream write mode.

Table-level parameters for stream writing

Stream writing parameters are table-level and must be added to the table parameters when creating a Hive table. These are primarily Filesystem table parameters. See Filesystem.
Additionally, some parameters are specific to stream writing to Hive:

Parameter	Required	Default value	Data type	Description
sink.partition-commit.policy.kind	Yes	metastore,success-file	String	The partition commit policy, which notifies downstream systems that a partition is finished writing and is ready to be read. We recommend setting this to metastore,success-file. metastore: Adds a partition to the metadata. Only Hive supports the metastore policy; file systems manage partitions through the directory structure. success-file: Adds a '_success' file to the directory. You can specify both policies simultaneously: 'metastore,success-file'. custom: Creates the commit policy using a specified class. Supports specifying multiple commit policies simultaneously: 'metastore,success-file'.

Parameter

Required

Default value

Data type

Description

sink.partition-commit.policy.kind

Yes

metastore,success-file

String

The partition commit policy, which notifies downstream systems that a partition is finished writing and is ready to be read. We recommend setting this to metastore,success-file.

metastore: Adds a partition to the metadata. Only Hive supports the metastore policy; file systems manage partitions through the directory structure.
success-file: Adds a '_success' file to the directory. You can specify both policies simultaneously: 'metastore,success-file'.
custom: Creates the commit policy using a specified class. Supports specifying multiple commit policies simultaneously: 'metastore,success-file'.

Sample code

The sample SQL uses default_hive as the Hive Catalog example.

Create a catalog

CREATE CATALOG default_hive WITH (
    'type' = 'hive',
    -- Path to the hive-site.xml file
    'hive-conf-dir' = '/opt/tiger/workdir'
);

Batch Read

SELECT * FROM default_hive.db.table WHERE `date`='${date}'

Batch Write

-- Insert new data into a non-partitioned table
INSERT INTO default_hive.db.mytable SELECT 'Tom', 25; 
-- Overwrite a non-partitioned table
INSERT OVERWRITE default_hive.db.mytable SELECT 'Tom', 25; 
-- Insert new data into a partitioned table
INSERT INTO default_hive.db.myparttable PARTITION (`date`='${date}') SELECT 'Tom', 25; 
-- Overwrite a partitioned table
INSERT OVERWRITE default_hive.db.myparttable PARTITION (`date`='${date}') SELECT 'Tom', 25;

Streaming Write

Create a Hive table and add parameters for streaming writes to Hive in the table properties:

CREATE TABLE db.hive_table (
  user_id STRING,
  order_amount DOUBLE
) PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (
  'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
  'sink.partition-commit.trigger'='partition-time',
  'sink.partition-commit.delay'='1 h',
  'sink.partition-commit.policy.kind'='metastore,success-file'
);

Flink SQL for streaming writes to Hive:

CREATE TABLE kafka_table (
  user_id STRING,
  order_amount DOUBLE,
  log_ts TIMESTAMP(3),
  WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND -- Declare a watermark on the TIMESTAMP column.
) WITH (...);

INSERT INTO TABLE default_hive.db.hive_table 
SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH')
FROM kafka_table;

Last updated: 2026.03.31 16:43:10

Streaming Computing Flink Version

Limitations #

Creating Tables #

Task Dependencies #

Dependency Files #

Custom Parameters #

Parameters #

Parameters for creating a catalog #

Job-level parameters for batch reading and writing #

Table-level parameters for stream writing #

Sample code #

Create a catalog #

Batch Read #

Batch Write #

Streaming Write #