pyflink.table.statement_set.StatementSet.add_insert

StatementSet.add_insert(target_path_or_descriptor: Union[str, pyflink.table.table_descriptor.TableDescriptor], table, overwrite: bool = False) → pyflink.table.statement_set.StatementSet

Adds a statement that the pipeline defined by the given Table object should be written to a table (backed by a DynamicTableSink) that was registered under the specified path or expressed via the given TableDescriptor.

1. When target_path_or_descriptor is a tale path:

   See the documentation of use_database() or use_catalog() for the rules on the path resolution.

2. When target_path_or_descriptor is a table descriptor:

   The given TableDescriptor is registered as an inline (i.e. anonymous) temporary catalog table (see create_temporary_table()).

   Then a statement is added to the statement set that inserts the Table object’s pipeline into that temporary table.

   This method allows to declare a Schema for the sink descriptor. The declaration is similar to a {@code CREATE TABLE} DDL in SQL and allows to:

   1. overwrite automatically derived columns with a custom DataType

   2. add metadata columns next to the physical columns

   3. declare a primary key

   It is possible to declare a schema without physical/regular columns. In this case, those columns will be automatically derived and implicitly put at the beginning of the schema declaration.

   Examples:

   >>> stmt_set = table_env.create_statement_set()
   >>> source_table = table_env.from_path("SourceTable")
   >>> sink_descriptor = TableDescriptor.for_connector("blackhole") \
   ...     .schema(Schema.new_builder()
   ...         .build()) \
   ...     .build()
   >>> stmt_set.add_insert(sink_descriptor, source_table)
   

   Note

   add_insert for a table descriptor (case 2.) was added from flink 1.14.0.

Parameters

• target_path_or_descriptor – The path of the registered TableSink or the descriptor describing the sink table into which data should be inserted to which the Table is written.

• table (pyflink.table.Table) – The Table to add.

• overwrite – Indicates whether the insert should overwrite existing data or not.

Returns

current StatementSet instance.

New in version 1.11.0.