INSERT Statement

    Java

    Single INSERT statement can be executed through the method of the TableEnvironment. The executeSql() method for INSERT statement will submit a Flink job immediately, and return a TableResult instance which associates the submitted job. Multiple INSERT statements can be executed through the addInsertSql() method of the StatementSet which can be created by the TableEnvironment.createStatementSet() method. The addInsertSql() method is a lazy execution, they will be executed only when StatementSet.execute() is invoked.

    The following examples show how to run a single INSERT statement in TableEnvironment, run multiple INSERT statements in StatementSet.

    Scala

    Single INSERT statement can be executed through the executeSql() method of the TableEnvironment. The executeSql() method for INSERT statement will submit a Flink job immediately, and return a TableResult instance which associates the submitted job. Multiple INSERT statements can be executed through the addInsertSql() method of the StatementSet which can be created by the TableEnvironment.createStatementSet() method. The addInsertSql() method is a lazy execution, they will be executed only when StatementSet.execute() is invoked.

    The following examples show how to run a single INSERT statement in TableEnvironment, run multiple INSERT statements in StatementSet.

    Python

    Single INSERT statement can be executed through the execute_sql() method of the TableEnvironment. The execute_sql() method for INSERT statement will submit a Flink job immediately, and return a TableResult instance which associates the submitted job. Multiple INSERT statements can be executed through the add_insert_sql() method of the StatementSet which can be created by the TableEnvironment.create_statement_set() method. The add_insert_sql() method is a lazy execution, they will be executed only when StatementSet.execute() is invoked.

    The following examples show how to run a single INSERT statement in TableEnvironment, run multiple INSERT statements in StatementSet.

    Single INSERT statement can be executed in SQL CLI.

    The following examples show how to run a single INSERT statement in SQL CLI.

    Java

    Scala

    1. val tEnv = TableEnvironment.create(...)
    3. // register a source table named "Orders" and a sink table named "RubberOrders"
    4. tEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
    5. tEnv.executeSql("CREATE TABLE RubberOrders(product STRING, amount INT) WITH (...)")
    7. // run a single INSERT query on the registered source table and emit the result to registered sink table
    8. val tableResult1 = tEnv.executeSql(
    9.   "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")
    10. // get job status through TableResult
    11. println(tableResult1.getJobClient().get().getJobStatus())
    13. //----------------------------------------------------------------------------
    14. // register another sink table named "GlassOrders" for multiple INSERT queries
    15. tEnv.executeSql("CREATE TABLE GlassOrders(product VARCHAR, amount INT) WITH (...)")
    17. // run multiple INSERT queries on the registered source table and emit the result to registered sink tables
    18. val stmtSet = tEnv.createStatementSet()
    19. // only single INSERT query can be accepted by `addInsertSql` method
    20. stmtSet.addInsertSql(
    21.   "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")
    22. stmtSet.addInsertSql(
    23.   "INSERT INTO GlassOrders SELECT product, amount FROM Orders WHERE product LIKE '%Glass%'")
    24. // execute all statements together
    25. val tableResult2 = stmtSet.execute()
    26. // get job status through TableResult
    27. println(tableResult2.getJobClient().get().getJobStatus())

    Python

    1. table_env = TableEnvironment.create(...)
    2. # register a source table named "Orders" and a sink table named "RubberOrders"
    3. table_env.execute_sql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
    4. table_env.execute_sql("CREATE TABLE RubberOrders(product STRING, amount INT) WITH (...)")
    6. # run a single INSERT query on the registered source table and emit the result to registered sink table
    7. table_result1 = table_env \
    8.     .execute_sql("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")
    9. # get job status through TableResult
    10. print(table_result1get_job_client().get_job_status())
    12. #----------------------------------------------------------------------------
    13. # register another sink table named "GlassOrders" for multiple INSERT queries
    14. table_env.execute_sql("CREATE TABLE GlassOrders(product VARCHAR, amount INT) WITH (...)")
    16. # run multiple INSERT queries on the registered source table and emit the result to registered sink tables
    17. stmt_set = table_env.create_statement_set()
    18. # only single INSERT query can be accepted by `add_insert_sql` method
    19. stmt_set \
    20.     .add_insert_sql("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")
    21. stmt_set \
    22.     .add_insert_sql("INSERT INTO GlassOrders SELECT product, amount FROM Orders WHERE product LIKE '%Glass%'")
    23. # execute all statements together
    24. table_result2 = stmt_set.execute()
    25. # get job status through TableResult
    26. print(table_result2.get_job_client().get_job_status())

    SQL CLI

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

    2. [EXECUTE] INSERT { INTO | OVERWRITE } [catalog_name.][db_name.]table_name [PARTITION part_spec] [column_list] select_statement
    4. part_spec:
    5.   (part_col_name1=val1 [, part_col_name2=val2, ...])
    7. column_list:
    8.   (col_name1 [, column_name2, ...])

    OVERWRITE

    PARTITION

    clause should contain static partition columns of this inserting.

    COLUMN LIST

    Given a table T(a INT, b INT, c INT), Flink supports INSERT INTO T(c, b) SELECT x, y FROM S. The expectation is that ‘x’ is written to column ‘c’ and ‘y’ is written to column ‘b’ and ‘a’ is set to NULL (assuming column ‘a’ is nullable).

    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. INSERT INTO country_page_view PARTITION (date='2019-8-30', country='China')
    7.   SELECT user, cnt FROM page_view_source;
    9. -- Key word EXECUTE can be added at the beginning of Insert to indicate explicitly that we are going to execute the statement,
    10. -- it is equivalent to Statement without the key word. 
    11. EXECUTE INSERT INTO country_page_view PARTITION (date='2019-8-30', country='China')
    12.   SELECT user, cnt FROM page_view_source;
    14. -- Appends rows into partition (date, country), where date is static partition with value '2019-8-30',
    15. -- country is dynamic partition whose value is dynamic determined by each row.
    16. INSERT INTO country_page_view PARTITION (date='2019-8-30')
    17.   SELECT user, cnt, country FROM page_view_source;
    19. -- Overwrites rows into static partition (date='2019-8-30', country='China')
    20. INSERT OVERWRITE country_page_view PARTITION (date='2019-8-30', country='China')
    21.   SELECT user, cnt FROM page_view_source;
    23. -- Overwrites rows into partition (date, country), where date is static partition with value '2019-8-30',
    24. -- country is dynamic partition whose value is dynamic determined by each row.
    25. INSERT OVERWRITE country_page_view PARTITION (date='2019-8-30')
    26.   SELECT user, cnt, country FROM page_view_source;
    28. -- Appends rows into the static partition (date='2019-8-30', country='China')
    29. -- the column cnt is set to NULL
    30. INSERT INTO country_page_view PARTITION (date='2019-8-30', country='China') (user)
    31.   SELECT user FROM page_view_source;

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

    OVERWRITE

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

    Examples

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

    The STATEMENT SET can be used to insert data into multiple tables in a statement.

    1. EXECUTE STATEMENT SET
    2. BEGIN
    3. insert_statement;
    4. ...
    5. insert_statement;
    6. END;

    Examples