# Evolve a Streaming Pipeline with Confluent Cloud for Apache Flink Confluent Cloud for Apache Flink® enables evolving streaming pipelines in place by using materialized tables — persistent objects that combine a table definition with a continuous query. In this guide, you create a materialized table with a simple query and then evolve it by adding a new column, without recreating the table or switching output topics. This guide uses the `examples.marketplace.orders` table, which is a read-only source of sample data available in every Flink environment. This guide shows the following steps: - [Step 1: Create a materialized table](#flink-sql-evolve-pipeline-create-mt) - [Step 2: Evolve by adding a column](#flink-sql-evolve-pipeline-evolve) - [Step 3: (Optional) Control reprocessing with START_MODE](#flink-sql-evolve-pipeline-start-mode) ## Prerequisites - Access to Confluent Cloud. - The FlinkDeveloper role or a higher role, such as EnvironmentAdmin. For more information, see [Grant Role-Based Access in Confluent Cloud for Apache Flink](../operate-and-deploy/flink-rbac.md#flink-rbac). - A provisioned Flink compute pool. ## Step 1: Create a materialized table 1. Log in to Confluent Cloud and navigate to your Flink workspace. 2. Create a materialized table that selects a subset of columns from the `examples.marketplace.orders` table. ```sql CREATE MATERIALIZED TABLE my_orders AS SELECT order_id, price FROM examples.marketplace.orders; ``` 3. Verify the materialized table was created by describing its schema. ```sql DESCRIBE my_orders; ``` Your output should resemble: ```none +----------+--------+------+-----+--------+-----------+ | name | type | null | key | extras | watermark | +----------+--------+------+-----+--------+-----------+ | order_id | STRING | TRUE | | | | | price | DOUBLE | TRUE | | | | +----------+--------+------+-----+--------+-----------+ ``` 4. Query the materialized table to verify data is flowing. ```sql SELECT * FROM my_orders; ``` ## Step 2: Evolve by adding a column Evolve the materialized table to include the `customer_id` column by using [CREATE OR ALTER MATERIALIZED TABLE](../reference/statements/create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table). 1. Run the following statement to add the `customer_id` column. ```sql CREATE OR ALTER MATERIALIZED TABLE my_orders AS SELECT order_id, customer_id, price FROM examples.marketplace.orders; ``` Flink stops the previous query and starts a new one with the updated projection, but the statement continues to write results to the same output topic. 2. Verify the updated schema. ```sql DESCRIBE my_orders; ``` Your output should now include the `customer_id` column: ```none +-------------+---------+------+-----+--------+-----------+ | name | type | null | key | extras | watermark | +-------------+---------+------+-----+--------+-----------+ | order_id | STRING | TRUE | | | | | customer_id | INT | TRUE | | | | | price | DOUBLE | TRUE | | | | +-------------+---------+------+-----+--------+-----------+ ``` 3. Query the materialized table to verify that the new column contains data. ```sql SELECT * FROM my_orders; ``` ## Step 3: (Optional) Control reprocessing with START_MODE By default, `CREATE OR ALTER` uses `RESUME_OR_FROM_BEGINNING`, which attempts to resume from the previous job’s position. You can control reprocessing behavior explicitly with the `START_MODE` clause. For example, to force a full reprocessing of all available historical data: ```sql CREATE OR ALTER MATERIALIZED TABLE my_orders START_MODE = FROM_BEGINNING AS SELECT order_id, customer_id, price FROM examples.marketplace.orders; ``` For the full list of `START_MODE` values, see [START_MODE](../reference/statements/create-or-alter-materialized-table.md#flink-sql-start-mode). ## Clean up Remove the materialized table when you are done. ```sql DROP MATERIALIZED TABLE IF EXISTS my_orders; ``` ## Related content - [CREATE MATERIALIZED TABLE](../reference/statements/create-materialized-table.md#flink-sql-create-materialized-table) - [CREATE OR ALTER MATERIALIZED TABLE](../reference/statements/create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table) - [DROP MATERIALIZED TABLE](../reference/statements/drop-materialized-table.md#flink-sql-drop-materialized-table) - [Materialized Tables concept](../concepts/materialized-tables.md#flink-sql-materialized-tables) - [Schema and Statement Evolution](../concepts/schema-statement-evolution.md#flink-sql-schema-and-statement-evolution) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).