Realtime Compute for Apache Flink:Manage Hive catalogs

Prerequisites

Limits

• Self-managed Hive metastores are supported.

• Hive 1.X, 2.1.X, and 2.2.X are not supported in Apache Flink 1.16 or later. Therefore, only Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 6.X supports Hive 1.X, 2.1.X, and 2.2.X.

• If DLF is used as the metastore for your Hive catalog, you can create non-Hive tables in the catalog only in Realtime Compute for Apache Flink that uses VVR 8.0.6 or later.

• You can use a Hive catalog to write data to OSS-HDFS only in Realtime Compute for Apache Flink that uses VVR 8.0.6 or later.

Configure Hive metadata

1. Establish a connection between a Hadoop cluster and the virtual private cloud (VPC) in which Realtime Compute for Apache Flink resides.

   You can use Alibaba Cloud DNS PrivateZone to connect a Hadoop cluster to the VPC in which Realtime Compute for Apache Flink resides. For more information, see Resolver. After the connection is established, Realtime Compute for Apache Flink can access the Hadoop cluster by using the configuration file of the Hadoop cluster.

2. Hive metastores or Alibaba Cloud DLF can be used as the metadata management center for Hive catalogs. The following section describes the related configurations.

   Hive MetaStore

   Check whether the setting of the hive.metastore.uris parameter in the Hive configuration file hive-site.xml meets the following requirements:

   <property>
       <name>hive.metastore.uris</name>
       <value>thrift://xx.yy.zz.mm:9083</value>
       <description>Thrift URI for the remote metastore. Used by metastore client to connect to remote metastore.</description>
    </property>

   xx.yy.zz.mm in the configuration file indicates the internal or public IP address of Hive.

   Note

   If you set the hive.metastore.uris parameter to the hostname of Hive, you must configure the Alibaba Cloud DNS service to parse the parameter value. Otherwise, the value of the hive.metastore.uris parameter fails to be parsed and the error message UnknownHostException is returned when Ververica Platform (VVP) remotely accesses Hive. For more information about how to configure the Alibaba Cloud DNS service, see Add a DNS record to a private zone.

   Alibaba Cloud DLF

   Add the following configurations to the Hive configuration file hive-site.xml to ensure that the Hive catalog can access DLF.

   Note

   If the hive-site.xml file contains the dlf.catalog.akMode configuration item, you must delete this configuration item. Otherwise, the Hive catalog cannot access DLF.

   <property>
     <name>hive.imetastoreclient.factory.class</name>
     <value>com.aliyun.datalake.metastore.hive2.DlfMetaStoreClientFactory</value>
   </property>
   <property>
     <name>dlf.catalog.uid</name>
     <value>${YOUR_DLF_CATALOG_UID}</value>
   </property>
   <property>
     <name>dlf.catalog.endpoint</name>
     <value>${YOUR_DLF_ENDPOINT}</value>
   </property>
   <property>
     <name>dlf.catalog.region</name>
     <value>${YOUR_DLF_CATALOG_REGION}</value>
   </property>
   <property>
     <name>dlf.catalog.accessKeyId</name>
     <value>${YOUR_ACCESS_KEY_ID}</value>
   </property>
   <property>
     <name>dlf.catalog.accessKeySecret</name>
     <value>${YOUR_ACCESS_KEY_SECRET}</value>
   </property>

   Parameter	Description	Remarks
   dlf.catalog.uid	The ID of your Alibaba Cloud account that is used to access DLF.	To obtain the ID of your Alibaba Cloud account, go to the Security Settings page.
   dlf.catalog.endpoint	The endpoint of the DLF service.	For more information, see Supported regions and endpoints. Note We recommend that you set dlf.endpoint to the VPC endpoint of DLF. For example, if you select the China (Hangzhou) region, set the dlf.endpoint parameter to dlf-vpc.cn-hangzhou.aliyuncs.com. If you want to access DLF across VPCs, follow the instructions that are described in How does Realtime Compute for Apache Flink access a service across VPCs?
   dlf.catalog.region	The ID of the region in which the DLF service is activated.	For more information, see Supported regions and endpoints. Note Make sure that the region you selected matches the endpoint you selected for dlf.endpoint.
   dlf.catalog.accessKeyI	The AccessKey ID of your Alibaba Cloud account.	For more information about how to obtain an AccessKey secret, see Obtain an AccessKey pair.
   dlf.catalog.accessKeySecret	The AccessKey secret of your Alibaba Cloud account.	For more information about how to obtain an AccessKey secret, see Obtain an AccessKey pair.

3. Hive catalogs allow you to store tables in OSS and OSS-HDFS. The following table describes the configurations.

   OSS

   Add the following configurations to the Hive configuration file hive-site.xml to ensure that the Hive catalog can access OSS.

   <property>
     <name>fs.oss.impl.disable.cache</name>
     <value>true</value>
   </property>
   <property>
     <name>fs.oss.impl</name>
     <value>org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem</value>
   </property>
   <property>
     <name>hive.metastore.warehouse.dir</name>
     <value>${YOUR_OSS_WAREHOUSE_DIR}</value>
   </property>
   <property>
     <name>fs.oss.endpoint</name>
     <value>${YOUR_OSS_ENDPOINT}</value>
   </property>
   <property>
     <name>fs.oss.accessKeyId</name>
     <value>${YOUR_ACCESS_KEY_ID}</value>
   </property>
   <property>
     <name>fs.oss.accessKeySecret</name>
     <value>${YOUR_ACCESS_KEY_SECRET}</value>
   </property>
   <property>
     <name>fs.defaultFS</name>
     <value>oss://${YOUR_OSS_BUCKET_DOMIN}</value>
   </property>

   The following table describes the parameters.

   Parameter	Description	Remarks
   hive.metastore.warehouse.dir	The directory in which table data is stored.	N/A.
   fs.oss.endpoint	The endpoint of OSS.	For more information, see Regions and endpoints.
   fs.oss.accessKeyId	The AccessKey ID of your Alibaba Cloud account.	For more information about how to obtain an AccessKey ID, see Obtain an AccessKey pair.
   fs.oss.accessKeySecret	The AccessKey secret of your Alibaba Cloud account.	For more information about how to obtain an AccessKey secret, see Obtain an AccessKey pair.
   fs.defaultFS	The file system in which table data is stored by default.	Make sure that the value of this parameter is the endpoint of the HDFS service that corresponds to the destination bucket, such as oss://oss-hdfs-bucket.cn-hangzhou.oss-dls.aliyuncs.com/.

   OSS-HDFS

   1. Add the following configurations to the Hive configuration file hive-site.xml to ensure that the Hive catalog can access OSS-HDFS.

      <property>
        <name>fs.jindo.impl</name>
        <value>com.aliyun.jindodata.jindo.JindoFileSystem</value>
      </property>
      <property>
        <name>hive.metastore.warehouse.dir</name>
        <value>${YOUR_OSS_WAREHOUSE_DIR}</value>
      </property>
      <property>
        <name>fs.oss.endpoint</name>
        <value>${YOUR_OSS_ENDPOINT}</value>
      </property>
      <property>
        <name>fs.oss.accessKeyId</name>
        <value>${YOUR_ACCESS_KEY_ID}</value>
      </property>
      <property>
        <name>fs.oss.accessKeySecret</name>
        <value>${YOUR_ACCESS_KEY_SECRET}</value>
      </property>
      <property>
        <name>fs.defaultFS</name>
        <value>oss://${YOUR_OSS_HDFS_BUCKET_DOMIN}</value>
      </property>

      Parameter	Description	Remarks
      hive.metastore.warehouse.dir	The directory in which table data is stored.	N/A.
      fs.oss.endpoint	The endpoint of OSS.	For more information, see Regions and endpoints.
      fs.oss.accessKeyId	The AccessKey ID of your Alibaba Cloud account.	For more information about how to obtain an AccessKey ID, see Obtain an AccessKey pair.
      fs.oss.accessKeySecret	The AccessKey secret of your Alibaba Cloud account.	For more information about how to obtain an AccessKey secret, see Obtain an AccessKey pair.
      fs.defaultFS	The file system in which table data is stored by default.	Make sure that the value of this parameter is the endpoint of the HDFS service that corresponds to the destination bucket, such as oss://oss-hdfs-bucket.cn-hangzhou.oss-dls.aliyuncs.com/.

   2. (Optional) If you want to read data from a Hive table stored in OSS-HDFS and the Hive table is in the Parquet format, add the following configuration items to Realtime Compute for Apache Flink.

      fs.oss.jindo.accessKeyId: ${YOUR_ACCESS_KEY_ID}
      fs.oss.jindo.accessKeySecret: ${YOUR_ACCESS_KEY_SECRET}
      fs.oss.jindo.endpoint: ${YOUR_JINODO_ENDPOINT}
      fs.oss.jindo.buckets: ${YOUR_JINDO_BUCKETS}

      For more information about the parameters, see Write data to OSS-HDFS.

   Note

   If your data is stored in Realtime Compute for Apache Flink, you do not need to perform the operations after this step. You can directly go to the Create a Hive Catalog step.

4. In the OSS console, create two directories in an OSS bucket and upload the Hive configuration file and Hadoop dependencies to these directories.

   1. Log on to the OSS console.

   2. In the left-side navigation pane, click Buckets.

   3. On the Buckets page, click the name of the desired bucket.

   4. Create a folder named ${hms} in the oss://${bucket}/artifacts/namespaces/${ns}/ path.

      For more information about how to create a directory in the OSS console, see Create directories. The following table describes the variables in the path in which you want to create a directory.

      Folder	Description
      ${bucket}	The name of the bucket that is used by your Realtime Compute for Apache Flink workspace.
      ${ns}	The name of the Realtime Compute for Apache Flink workspace for which you want to create a Hive catalog.
      ${hms}	The name of the folder to create. We recommend that you use the same name for this folder and the Hive catalog that you will create in subsequent steps.

      Note

      After you create a Realtime Compute for Apache Flink workspace, Realtime Compute for Apache Flink automatically creates the /artifacts/namespaces/${ns}/ directory in the specified bucket to store data, such as JAR packages. If you do not find the directory in the OSS console, you must manually upload a file to create the directory on the Artifacts page in the development console of Realtime Compute for Apache Flink.

   5. Create a directory named hive-conf-dir and a directory named hadoop-conf-dir in the oss://${bucket}/artifacts/namespaces/${ns}/${hms} path. For more information about how to create a directory in the OSS console, see Create directories.

      The following examples describe the files that are stored in the hive-conf-dir and hadoop-conf-dir directories:

      • oss://${bucket}/artifacts/namespaces/${ns}/${hms}/hive-conf-dir/ is used to store the Hive configuration file named hive-site.xml.

      • oss://${bucket}/artifacts/namespaces/${ns}/${hms}/hadoop-conf-dir/ is used to store the Hadoop configuration files, such as core-site.xml, hdfs-site.xml, yarn-site.xml, and mapred-site.xml.

      After the directories are created, you can choose Files > Projects in the left-side navigation pane in the OSS console to view the new directories and files, and then copy the OSS URL.

   6. Upload the Hive configuration file hive-site.xml to the hive-conf-dir directory. For more information about how to upload a file, see Upload objects.

   7. Upload the following configuration files to the hadoop-conf-dir directory. For more information about how to upload a file, see Upload objects.

      • hive-site.xml

      • core-site.xml

      • hdfs-site.xml

      • mapred-site.xml

      • Other required files, such as the compressed packages used by Hive deployments

Create a Hive catalog

After you configure Hive metadata, you can create a Hive catalog on the UI or by executing an SQL statement. We recommend that you create a Hive catalog on the UI.

Use a Hive catalog

-- Create a table named flink_hive_test in the flinkhive database under the flinkexporthive catalog. 
CREATE TABLE `flinkexporthive`.`flinkhive`.`flink_hive_test` (
  id INT,
  name STRING
) WITH (
  'connector' = 'hive'
);

-- Add a column to the Hive table. 
ALTER TABLE `${catalog_name}`.`${db_name}`.`${table_name}` 
ADD column type-column;

-- Drop a column from the Hive table. 
ALTER TABLE `${catalog_name}`.`${db_name}`.`${table_name}` 
DROP column;

-- Add the color field to the Hive table. 
ALTER TABLE `flinkexporthive`.`flinkhive`.`flink_hive_test` 
ADD color STRING;

-- Drop the color field from the Hive table. 
ALTER TABLE `flinkexporthive`.`flinkhive`.`flink_hive_test` 
DROP color;

INSERT INTO ${other_sink_table}
SELECT ...
FROM `${catalog_name}`.`${db_name}`.`${table_name}`;

INSERT INTO `${catalog_name}`.`${db_name}`.`${table_name}`
SELECT ... 
FROM ${other_source_table};

-- Drop the Hive table. 
DROP TABLE `${catalog_name}`.`${db_name}`.`${table_name}`;

-- Drop the Hive table. 
DROP TABLE `flinkexporthive`.`flinkhive`.`flink_hive_test`;

View a Hive catalog

1. Go to the Catalogs page.

   1. Log on to the Realtime Compute for Apache Flink console.

   2. Find the workspace that you want to manage and click Console in the Actions column.

   3. Click Catalogs.

2. On the Catalog List page, find the desired catalog and view the Name and Type columns of the catalog.

   Note

   If you want to view the databases and tables in the catalog, click View in the Actions column.

Drop a Hive catalog