Hive Dialect

    Flink currently supports two SQL dialects: default and hive. You need to switch to Hive dialect before you can write in Hive syntax. The following describes how to set dialect with SQL Client and Table API. Also notice that you can dynamically switch dialect for each statement you execute. There’s no need to restart a session to use a different dialect.

    SQL dialect can be specified via the table.sql-dialect property. Therefore you can set the initial dialect to use in the configuration section of the yaml file for your SQL Client.

    You can also set the dialect after the SQL Client has launched.

    2. Flink SQL> SET 'table.sql-dialect' = 'hive'; -- to use hive dialect
    3. [INFO] Session property has been set.
    5. Flink SQL> SET 'table.sql-dialect' = 'default'; -- to use default dialect
    6. [INFO] Session property has been set.

    Table API

    You can set dialect for your TableEnvironment with Table API.

    2. EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
    3. TableEnvironment tableEnv = TableEnvironment.create(settings);
    4. // to use hive dialect
    5. tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
    6. // to use default dialect
    7. tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);

    Python

    1. from pyflink.table import *
    3. settings = EnvironmentSettings.in_batch_mode()
    4. t_env = TableEnvironment.create(settings)
    6. # to use hive dialect
    7. t_env.get_config().set_sql_dialect(SqlDialect.HIVE)
    8. # to use default dialect
    9. t_env.get_config().set_sql_dialect(SqlDialect.DEFAULT)

    This section lists the supported DDLs with the Hive dialect. We’ll mainly focus on the syntax here. You can refer to for the semantics of each DDL statement.

    Show

      DATABASE

      Show

      1. SHOW DATABASES;
      2. SHOW CURRENT DATABASE;

      Create

      1. CREATE (DATABASE|SCHEMA) [IF NOT EXISTS] database_name
      2.   [COMMENT database_comment]
      3.   [LOCATION fs_path]
      4.   [WITH DBPROPERTIES (property_name=property_value, ...)];

      Alter

      Update Properties
      1. ALTER (DATABASE|SCHEMA) database_name SET DBPROPERTIES (property_name=property_value, ...);
      Update Owner
      1. ALTER (DATABASE|SCHEMA) database_name SET OWNER [USER|ROLE] user_or_role;
      Update Location
      1. ALTER (DATABASE|SCHEMA) database_name SET LOCATION fs_path;

      Drop

      1. DROP (DATABASE|SCHEMA) [IF EXISTS] database_name [RESTRICT|CASCADE];

      Use

      Show

      1. SHOW TABLES;

      Create

      1. CREATE [EXTERNAL] TABLE [IF NOT EXISTS] table_name
      2.   [(col_name data_type [column_constraint] [COMMENT col_comment], ... [table_constraint])]
      3.   [COMMENT table_comment]
      4.   [PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)]
      5.   [
      6.     [ROW FORMAT row_format]
      7.     [STORED AS file_format]
      8.   ]
      9.   [LOCATION fs_path]
      10.   [TBLPROPERTIES (property_name=property_value, ...)]
      12. row_format:
      13.   : DELIMITED [FIELDS TERMINATED BY char [ESCAPED BY char]] [COLLECTION ITEMS TERMINATED BY char]
      14.       [MAP KEYS TERMINATED BY char] [LINES TERMINATED BY char]
      15.       [NULL DEFINED AS char]
      16.   | SERDE serde_name [WITH SERDEPROPERTIES (property_name=property_value, ...)]
      18. file_format:
      19.   : SEQUENCEFILE
      20.   | TEXTFILE
      21.   | RCFILE
      22.   | ORC
      23.   | PARQUET
      24.   | AVRO
      25.   | INPUTFORMAT input_format_classname OUTPUTFORMAT output_format_classname
      27. column_constraint:
      28.   : NOT NULL [[ENABLE|DISABLE] [VALIDATE|NOVALIDATE] [RELY|NORELY]]
      30. table_constraint:
      31.   : [CONSTRAINT constraint_name] PRIMARY KEY (col_name, ...) [[ENABLE|DISABLE] [VALIDATE|NOVALIDATE] [RELY|NORELY]]

      Alter

      Rename
      1. ALTER TABLE table_name RENAME TO new_table_name;
      Update Properties
      1. ALTER TABLE table_name SET TBLPROPERTIES (property_name = property_value, property_name = property_value, ... );
      Update Location
      1. ALTER TABLE table_name [PARTITION partition_spec] SET LOCATION fs_path;

      The partition_spec, if present, needs to be a full spec, i.e. has values for all partition columns. And when it’s present, the operation will be applied to the corresponding partition instead of the table.

      Update File Format
      1. ALTER TABLE table_name [PARTITION partition_spec] SET FILEFORMAT file_format;

      The partition_spec, if present, needs to be a full spec, i.e. has values for all partition columns. And when it’s present, the operation will be applied to the corresponding partition instead of the table.

      Update SerDe Properties
      1. ALTER TABLE table_name [PARTITION partition_spec] SET SERDE serde_class_name [WITH SERDEPROPERTIES serde_properties];
      3. ALTER TABLE table_name [PARTITION partition_spec] SET SERDEPROPERTIES serde_properties;
      5. serde_properties:
      6.   : (property_name = property_value, property_name = property_value, ... )
      Add Partitions
      1. ALTER TABLE table_name ADD [IF NOT EXISTS] (PARTITION partition_spec [LOCATION fs_path])+;
      Drop Partitions
      1. ALTER TABLE table_name DROP [IF EXISTS] PARTITION partition_spec[, PARTITION partition_spec, ...];
      Add/Replace Columns
      1. ALTER TABLE table_name
      2.   ADD|REPLACE COLUMNS (col_name data_type [COMMENT col_comment], ...)
      3.   [CASCADE|RESTRICT]
      Change Column

      Drop

      1. DROP TABLE [IF EXISTS] table_name;

      VIEW

      Create

      1. CREATE VIEW [IF NOT EXISTS] view_name [(column_name, ...) ]
      2.   [COMMENT view_comment]
      3.   [TBLPROPERTIES (property_name = property_value, ...)]
      4.   AS SELECT ...;

      Alter

      Rename
      1. ALTER VIEW view_name RENAME TO new_view_name;
      Update Properties
      1. ALTER VIEW view_name SET TBLPROPERTIES (property_name = property_value, ... );
      Update As Select

        Drop

        1. DROP VIEW [IF EXISTS] view_name;

        Show

        1. SHOW FUNCTIONS;

        Create

        Drop

        1. DROP FUNCTION [IF EXISTS] function_name;

        Hive dialect supports a commonly-used subset of Hive’s and DQL. The following lists some examples of HiveQL supported by the Hive dialect.

        In order to have better syntax and semantic compatibility, it’s highly recommended to use and place it first in the module list, so that Hive built-in functions can be picked up during function resolution.

        Hive dialect no longer supports Flink SQL queries. Please switch to default dialect if you’d like to write in Flink syntax.

        Following is an example of using hive dialect to run some queries.

        1. Flink SQL> create catalog myhive with ('type' = 'hive', 'hive-conf-dir' = '/opt/hive-conf');
        2. [INFO] Execute statement succeed.
        4. Flink SQL> use catalog myhive;
        5. [INFO] Execute statement succeed.
        7. Flink SQL> load module hive;
        8. [INFO] Execute statement succeed.
        10. Flink SQL> use modules hive,core;
        11. [INFO] Execute statement succeed.
        13. Flink SQL> set table.sql-dialect=hive;
        14. [INFO] Session property has been set.
        16. Flink SQL> select explode(array(1,2,3)); -- call hive udtf
        17. +-----+
        18. | col |
        19. +-----+
        20. |   1 |
        21. |   2 |
        22. |   3 |
        23. +-----+
        24. 3 rows in set
        26. Flink SQL> create table tbl (key int,value string);
        27. [INFO] Execute statement succeed.
        29. Flink SQL> insert overwrite table tbl values (5,'e'),(1,'a'),(1,'a'),(3,'c'),(2,'b'),(3,'c'),(3,'c'),(4,'d');
        30. [INFO] Submitting SQL update statement to the cluster...
        31. [INFO] SQL update statement has been successfully submitted to the cluster:
        33. Flink SQL> select * from tbl cluster by key; -- run cluster by
        34. 2021-04-22 16:13:57,005 INFO  org.apache.hadoop.mapred.FileInputFormat                     [] - Total input paths to process : 1
        35. +-----+-------+
        36. | key | value |
        37. +-----+-------+
        38. |   1 |     a |
        39. |   1 |     a |
        40. |   5 |     e |
        41. |   2 |     b |
        42. |   3 |     c |
        43. |   3 |     c |
        44. |   3 |     c |
        45. |   4 |     d |
        47. 8 rows in set
        • Hive dialect should only be used to process Hive meta objects, and requires the current catalog to be a .
        • Hive dialect only supports 2-part identifiers, so you can’t specify catalog for an identifier.
        • While all Hive versions support the same syntax, whether a specific feature is available still depends on the Hive version you use. For example, updating database location is only supported in Hive-2.4.0 or later.