pyflink.table.table_environment.StreamTableEnvironment.from_changelog_stream

StreamTableEnvironment.from_changelog_stream(data_stream: pyflink.datastream.data_stream.DataStream, schema: Optional[pyflink.table.schema.Schema] = None, changelog_mode: Optional[pyflink.table.changelog_mode.ChangelogMode] = None) → pyflink.table.table.Table

Converts the given DataStream of changelog entries into a Table.

Compared to from_data_stream(), this method consumes instances of Row and evaluates the RowKind flag that is contained in every record during runtime. The runtime behavior is similar to that of a DynamicTableSource.

If you don’t specify the changelog_mode, the changelog containing all kinds of changes (enumerated in RowKind) as the default ChangelogMode.

Column names and types of the Table are automatically derived from the TypeInformation of the DataStream. If the outermost record’s TypeInformation is a CompositeType, it will be flattened in the first level. Composite nested fields will not be accessible.

By default, the stream record’s timestamp and watermarks are not propagated unless explicitly declared.

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

1. enrich or overwrite automatically derived columns with a custom DataType

2. reorder columns

3. add computed or metadata columns next to the physical columns

4. access a stream record’s timestamp

5. declare a watermark strategy or propagate the DataStream watermarks

6. declare a primary key

See from_data_stream() for more information and examples of how to declare a Schema.

Parameters

• data_stream – The changelog stream of Row.

• schema – The customized schema for the final table.

• changelog_mode – The expected kinds of changes in the incoming changelog.

Returns

The converted Table.