Source Table

Introduction

Apache Hive has established itself as a focal point of the data warehousing ecosystem. It serves as not only a SQL engine for big data analytics and ETL, but also a data management platform, where data is discovered, defined, and evolved.

Flink offers a two-fold integration with Hive. The first is to leverage Hive's Metastore as a persistent catalog. The second is to offer Flink as an alternative engine for reading and writing Hive tables. Overview | Apache Flink

Starting from 1.11.0, Flink allows users to write SQL statements in Hive syntax when Hive dialect is used. By providing compatibility with Hive syntax, we aim to improve the interoperability with Hive and reduce the scenarios when users need to switch between Flink and Hive in order to execute different statements. For details, see Apache Flink Hive Dialect.

Using the HiveCatalog, Apache Flink can be used for unified BATCH and STREAM processing of Apache Hive Tables. This means Flink can be used as a more performant alternative to Hive's batch engine, or to continuously read and write data into and out of Hive tables to power real-time data warehousing applications. Apache Flink Hive Read & Write

Function

This section describes how to use Flink to read and write Hive tables, the definition of the Hive source table, parameters used for creating the source table, and sample code. For details, see Apache Flink Hive Read & Write.

Flink supports reading data from Hive in both BATCH and STREAMING modes. When running as a BATCH application, Flink will execute its query over the state of the table at the point in time when the query is executed. STREAMING reads will continuously monitor the table and incrementally fetch new data as it is made available. Flink will read tables as bounded by default.

STREAMING reads support consuming both partitioned and non-partitioned tables. For partitioned tables, Flink will monitor the generation of new partitions, and read them incrementally when available. For non-partitioned tables, Flink will monitor the generation of new files in the folder and read new files incrementally.

Prerequisites

Caveats

• When you create a Flink OpenSource SQL job, set Flink Version to 1.15 in the Running Parameters tab. Select Save Job Log, and specify the OBS bucket for saving job logs.
• For details about how to use data types, see Format.
• Flink 1.15 currently only supports creating OBS tables and DLI lakehouse tables using Hive syntax, which is supported by Hive dialect DDL statements.
  • To create an OBS table using Hive syntax:
    • For the default dialect, set hive.is-external to true in the with properties.
    • For the Hive dialect, use the EXTERNAL keyword in the create table statement.
  • To create a DLI lakehouse table using Hive syntax:
    • For the Hive dialect, add 'is_lakehouse'='true' to the table properties.
• Enable checkpointing.
• You are advised to switch to Hive dialect to create Hive-compatible tables. If you want to create Hive-compatible tables with default dialect, make sure to set 'connector'='hive' in your table properties, otherwise a table is considered generic by default in HiveCatalog. Note that the connector property is not required if you use Hive dialect.
• Monitor strategy is to scan all directories/files currently in the location path. Many partitions may cause performance degradation.
• Streaming reads for non-partitioned tables requires that each file be written atomically into the target directory.
• Streaming reading for partitioned tables requires that each partition should be added atomically in the view of hive metastore. If not, new data added to an existing partition will be consumed.
• Streaming reads do not support watermark grammar in Flink DDL. These tables cannot be used for window operators.

Syntax

 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

CREATE EXTERNAL TABLE [IF NOT EXISTS] table_name
  [(col_name data_type [column_constraint] [COMMENT col_comment], ... [table_constraint])]
  [COMMENT table_comment]
  [PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)]
  [
    [ROW FORMAT row_format]
    [STORED AS file_format]
  ]
  [LOCATION obs_path]
  [TBLPROPERTIES (property_name=property_value, ...)]

row_format:
  : DELIMITED [FIELDS TERMINATED BY char [ESCAPED BY char]] [COLLECTION ITEMS TERMINATED BY char]
      [MAP KEYS TERMINATED BY char] [LINES TERMINATED BY char]
      [NULL DEFINED AS char]
  | SERDE serde_name [WITH SERDEPROPERTIES (property_name=property_value, ...)]

file_format:
  : SEQUENCEFILE
  | TEXTFILE
  | RCFILE
  | ORC
  | PARQUET
  | AVRO
  | INPUTFORMAT input_format_classname OUTPUTFORMAT output_format_classname

column_constraint:
  : NOT NULL [[ENABLE|DISABLE] [VALIDATE|NOVALIDATE] [RELY|NORELY]]

table_constraint:
  : [CONSTRAINT constraint_name] PRIMARY KEY (col_name, ...) [[ENABLE|DISABLE] [VALIDATE|NOVALIDATE] [RELY|NORELY]]

Parameter Description

For the semantics of each DDL statement, see Creating an OBS Table Using the Hive Syntax and Hive documentation.

• Source Parallelism Inference

  By default, Flink infers the hive source parallelism based on the number of splits, and the number of splits is based on the number of files and the number of blocks in the files.

  Flink allows you to flexibly configure the policy of parallelism inference. You can configure the following parameters in TableConfig (note that these parameters affect all sources of the job):

  Key	Default	Type	Description
  table.exec.hive.infer-source-parallelism	true	Boolean	If it is true, source parallelism is inferred according to splits number. If it is false, parallelism of source is set by config.
  table.exec.hive.infer-source-parallelism.max	1000	Integer	Sets max infer parallelism for source operator.

• Load Partition Splits

  Multi-thread is used to split hive's partitions. You can use table.exec.hive.load-partition-splits.thread-num to configure the thread number. The default value is 3 and the configured value should be greater than 0.

  Key	Default	Type	Description
  table.exec.hive.load-partition-splits.thread-num	3	Integer	The configured value should be greater than 0.

  SQL hints can be used to apply configurations to a Hive table without changing its definition in the Hive metastore. See Hints | Apache Flink.

• Vectorized Optimization upon Read

  Flink will automatically used vectorized reads of Hive tables when the following conditions are met:

  • Format: ORC or Parquet.
  • Columns without complex data type, like hive types: List, Map, Struct, Union.
    This feature is enabled by default. It may be disabled with the following configuration.

    table.exec.hive.fallback-mapred-reader=true

Example

1. Create an OBS table in Hive syntax using Spark SQL and insert 10 data records. Simulate the data source.

   CREATE TABLE IF NOT EXISTS demo.student(
     name STRING, 
    score DOUBLE) 
   PARTITIONED BY (classNo INT) 
   STORED AS PARQUET 
   LOCATION 'obs://demo/spark.db/student';
   
   INSERT INTO demo.student PARTITION(classNo=1) VALUES ('Alice', 90.0), ('Bob', 80.0), ('Charlie', 70.0), ('David', 60.0), ('Eve', 50.0), ('Frank', 40.0), ('Grace', 30.0), ('Hank', 20.0), ('Ivy', 10.0), ('Jack', 0.0);

2. Demonstrate batch processing using Flink SQL to read data from the Hive syntax OBS table demo.student in batch mode and print it out. Checkpointing is required.

   CREATE CATALOG myhive WITH (
       'type' = 'hive',
       'default-database' = 'demo',
        'hive-conf-dir' = '/opt/flink/conf'
   );
   
   USE CATALOG myhive;
   
   create table if not exists print (
       name STRING, 
       score DOUBLE, 
       classNo INT)
   with ('connector' = 'print');
   
   insert into print
   select * from student;

   Result (out log of TaskManager):

   +I[Alice, 90.0, 1]
   +I[Bob, 80.0, 1]
   +I[Charlie, 70.0, 1]
   +I[David, 60.0, 1]
   +I[Eve, 50.0, 1]
   +I[Frank, 40.0, 1]
   +I[Grace, 30.0, 1]
   +I[Hank, 20.0, 1]
   +I[Ivy, 10.0, 1]
   +I[Jack, 0.0, 1]

3. Demonstrate stream processing by using Flink SQL to read data from the Hive syntax OBS table demo.student in stream mode and print it out.

   CREATE CATALOG myhive WITH (
       'type' = 'hive' ,
       'default-database' = 'demo',
        'hive-conf-dir' = '/opt/flink/conf'
   );
   
   USE CATALOG myhive;
   
   create table if not exists print (
       name STRING, 
       score DOUBLE, 
       classNo INT)
   with ('connector' = 'print');
   
   insert into print
   select * from student /*+ OPTIONS('streaming-source.enable' = 'true', 'streaming-source.monitor-interval' = '3 m') */;