The Hive connector provides read and write access to Hive data sources. By using the Hive Catalog, Flink can perform unified batch and stream processing on Hive tables. This means Flink can be an alternative to the Hive batch engine or stream data into Hive tables to support real-time Data Warehouse applications.

1. Environment preparation #

1.1 Activate Serverless Flink and E-MapReduce #

Make sure you have:

1. Activated the Flink product and can create Flink SQL jobs in job development.
2. Purchased pay-as-you-go or subscription resource pools in the Resource Management - Resource Pools module and can successfully submit Flink jobs.
3. Activated the E-MapReduce product service. For instructions, see Create an EMR on ECS cluster.
   1. Note: EMR clusters must have the Paimon component installed; otherwise, you cannot use Hive and Spark to access Paimon data.

1.2 Product Requirements #

• The Flink version must be 1.16 or higher. The differences in supported features are as follows:
  • Flink 1.16: Supports both Kerberos-authenticated and non-Kerberos-authenticated clusters
  • Flink 1.17: Only supports non-Kerberos-authenticated clusters

2. Preparing Cluster Dependency Files #

2.1 File Checklist #

Prerequisites: Flink requires access to the E-MapReduce cluster. You must obtain the following files.

• hive-site.xml
• hdfs-site.xml
• Keytab file (Optional, required for Kerberos)
• Krb5.conf (Optional, required for Kerberos)

2.2 Log in to the EMR master node #

The configuration files can be obtained as follows: In the E-MapReduce product cluster, go to Cluster Details - Node Management - MasterGroup and select the first Master node:
Remotely connect to the specified Master server via the console or a public SSH connection, and access it through the Terminal console

Note: If you connect via public SSH, you need to bind a public IP address to the ECS instance in advance

2.3 hive-site.xml #

You can obtain it with the following command:

cd /etc/emr/hive/conf
env | grep HIVE_CONF_DIR

Copy the hive-site.xml file in the figure below to your local machine using the SCP command, or use the cat command to view the content:

2.4 hdfs-site.xml #

The procedure for hdfs-site.xml is the same as for hive-site.xml

Additionally, as some Hive table formats involve YARN classes, you may need to copy the yarn.resourcemanager.principal configuration from yarn-site.xml to hdfs-site.xml.
The content is similar to the following:

<property>
    <name>yarn.resourcemanager.principal</name>
    <value>yarn/_HOST@6C5B5406ADDF347BB8C9.EMR.COM</value>
</property>

2.5 keytab (Optional, required for Kerberos) #

On the EMR cluster overview page, create a user in User Management and then download the keytab.

2.6 krb5.conf (Optional, required for Kerberos) #

The krb5.conf file is located at /etc/krb5.conf. Copy it directly using scp or copy its text content via cat.

3. Create a Data Directory #

3.1 Create a Hive Database and Table #

Log in to the EMR master node and run the following commands in the Hive terminal to create a data table:

-- Create a Hive database.
create database flink_test;
use flink_test;

-- Create a Hive test table
create table flink_table (`a` int, `b` string);

3.2 Create a Data Directory #

Step 1: Click the Create Catalog button, select EMR Hive as the catalog type, and proceed to the next step.

Step 2: Enter the catalog name, select the EMR version and cluster name, and upload the hive-site.xml and hdfs-site.xml files. For a Kerberos cluster, you must also upload the keytab and krb5.conf files.

Step 3: The catalog is created successfully. EMR-Hive does not currently support metadata synchronization, so you cannot view the structure of all Hive databases and tables.

3.2 Referencing a Data Catalog in Flink SQL #

Step 1: In Flink SQL, you can directly access data tables using the three-part catalog.database.table format. For example, here is how to write to an LAS table:

create table orders (
    a int,
    b string
) WITH (
    'connector' = 'datagen',
    'rows-per-second'='1'
);


insert into `emr_hive`.`flink_test`.`flink_table`
select a , b from orders

Step 2: Save, publish, and launch the SQL job.
Step 3: Start the job and verify that the data was written successfully.
You can monitor the data writing status in Hive.

hive> use flink_test;
OK
Time taken: 0.088 seconds
hive> select * from flink_table limit 10;
OK
...
Time taken: 1.305 seconds, Fetched: 10 row(s)

3. Common Issues #

3.1 Data Catalog Synchronization Failure #

Network connectivity tests fail when creating a data catalog, resulting in a data catalog synchronization failure.
Cause: EMR-Hive does not currently support metadata synchronization, so you cannot view database and table information on the UI.

3.2 Unable to Access the Kerberos Cluster #

Symptom: You cannot access the Hive Metastore of a Kerberos cluster when using Flink 1.17, causing write tasks to fail.
Cause: Flink 1.17 does not currently support Kerberos clusters. Use Flink 1.16 instead.

3.3 EMR-Hive Cannot Validate SQL Correctness #

Symptom: The following error occurs when you click Validate or Go-live.
Cause: EMR-Hive does not currently support features such as quick validation and deep validation. Skip the deep check before going live.