INSERT Statement

INSERT statements are used to add rows to a table.

Run an INSERT statement

INSERT statements are specified with the sqlUpdate() method of the TableEnvironment or executed in SQL CLI. The method sqlUpdate() for INSERT statements is a lazy execution, they will be executed only when TableEnvironment.execute(jobName) is invoked.

The following examples show how to run an INSERT statement in TableEnvironment and in SQL CLI.

  1. EnvironmentSettings settings = EnvironmentSettings.newInstance()...
  2. TableEnvironment tEnv = TableEnvironment.create(settings);
  4. // register a source table named "Orders" and a sink table named "RubberOrders"
  5. tEnv.sqlUpdate("CREATE TABLE Orders (`user` BIGINT, product VARCHAR, amount INT) WITH (...)");
  6. tEnv.sqlUpdate("CREATE TABLE RubberOrders(product VARCHAR, amount INT) WITH (...)");
  8. // run a SQL update query on the registered source table and emit the result to registered sink table
  9. tEnv.sqlUpdate(
  10.   "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");
  1. val settings = EnvironmentSettings.newInstance()...
  2. val tEnv = TableEnvironment.create(settings)
  4. // register a source table named "Orders" and a sink table named "RubberOrders"
  5. tEnv.sqlUpdate("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
  6. tEnv.sqlUpdate("CREATE TABLE RubberOrders(product STRING, amount INT) WITH (...)")
  8. // run a SQL update query on the registered source table and emit the result to registered sink table
  9. tEnv.sqlUpdate(
  10.   "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")
  1. settings = EnvironmentSettings.newInstance()...
  2. table_env = TableEnvironment.create(settings)
  4. # register a source table named "Orders" and a sink table named "RubberOrders"
  5. table_env.sqlUpdate("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
  6. table_env.sqlUpdate("CREATE TABLE RubberOrders(product STRING, amount INT) WITH (...)")
  8. # run a SQL update query on the registered source table and emit the result to registered sink table
  9. table_env \
  10.     .sqlUpdate("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")
  1. Flink SQL> CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...);
  2. [INFO] Table has been created.
  4. Flink SQL> CREATE TABLE RubberOrders(product STRING, amount INT) WITH (...);
  6. Flink SQL> SHOW TABLES;
  7. Orders
  8. RubberOrders
  10. Flink SQL> INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%';
  11. [INFO] Submitting SQL update statement to the cluster...
  12. [INFO] Table update statement has been successfully submitted to the cluster:

Insert from select queries

Query Results can be inserted into tables by using the insert clause.

Syntax

  1. INSERT { INTO | OVERWRITE } [catalog_name.][db_name.]table_name [PARTITION part_spec] select_statement
  3. part_spec:
  4.   (part_col_name1=val1 [, part_col_name2=val2, ...])

OVERWRITE

INSERT OVERWRITE will overwrite any existing data in the table or partition. Otherwise, new data is appended.

PARTITION

PARTITION clause should contain static partition columns of this inserting.

Examples

  1. -- Creates a partitioned table
  2. CREATE TABLE country_page_view (user STRING, cnt INT, date STRING, country STRING)
  3. PARTITIONED BY (date, country)
  4. WITH (...)
  6. -- Appends rows into the static partition (date='2019-8-30', country='China')
  7. INSERT INTO country_page_view PARTITION (date='2019-8-30', country='China')
  8.   SELECT user, cnt FROM page_view_source;
  10. -- Appends rows into partition (date, country), where date is static partition with value '2019-8-30',
  11. -- country is dynamic partition whose value is dynamic determined by each row.
  12. INSERT INTO country_page_view PARTITION (date='2019-8-30')
  13.   SELECT user, cnt, country FROM page_view_source;
  15. -- Overwrites rows into static partition (date='2019-8-30', country='China')
  16. INSERT OVERWRITE country_page_view PARTITION (date='2019-8-30', country='China')
  17.   SELECT user, cnt FROM page_view_source;
  19. -- Overwrites rows into partition (date, country), where date is static partition with value '2019-8-30',
  20. -- country is dynamic partition whose value is dynamic determined by each row.
  21. INSERT OVERWRITE country_page_view PARTITION (date='2019-8-30')
  22.   SELECT user, cnt, country FROM page_view_source;

Insert values into tables

The INSERT…VALUES statement can be used to insert data into tables directly from SQL.

Syntax

  1. INSERT { INTO | OVERWRITE } [catalog_name.][db_name.]table_name VALUES values_row [, values_row ...]
  3. values_row:
  4.     : (val1 [, val2, ...])

OVERWRITE

INSERT OVERWRITE will overwrite any existing data in the table. Otherwise, new data is appended.

Examples

  1. CREATE TABLE students (name STRING, age INT, gpa DECIMAL(3, 2)) WITH (...);
  3. INSERT INTO students
  4.   VALUES ('fred flintstone', 35, 1.28), ('barney rubble', 32, 2.32);