# Quick Start: Build a Streamhouse on Confluent Cloud Build a small Streamhouse for River Air, a fictional airline, entirely in Confluent Cloud. When an inbound flight runs late, you keep one current state for every connecting passenger, serve it as Lightning Tables, let an AI agent rebook the passengers who miss their connections, and answer questions about history from the same data. ![Stream Lineage view of the finished quick start: Flink statements feed passenger_state, which the rebooking agent reads.](streamhouse/images/streamhouse-quick-start-lineage.png) This quick start takes about 30 minutes. You don’t need any other accounts or infrastructure: the AI model is a managed model that runs in Confluent Cloud, and Tableflow stores the history in Confluent-managed storage. The resources that you create in this quick start incur Confluent Cloud charges. To stop the charges when you’re done, follow the steps in [Step 8: Clean up](#streamhouse-quick-start-step-8). In this quick start, you perform the following steps: - [Step 1: Create an environment, a cluster, and a compute pool](#streamhouse-quick-start-step-1) - [Step 2: Create the source tables](#streamhouse-quick-start-step-2) - [Step 3: Keep one current state with a materialized table](#streamhouse-quick-start-step-3) - [Step 4: Delay the inbound flight](#streamhouse-quick-start-step-4) - [Step 5: Serve the current state with Lightning Tables](#streamhouse-quick-start-step-5) - [Step 6: Ground a rebooking agent in the current state](#streamhouse-quick-start-step-6) - [Step 7: Answer questions with history](#streamhouse-quick-start-step-7) - [Step 8: Clean up](#streamhouse-quick-start-step-8) ## Prerequisites - Access to Confluent Cloud, with permission to create an environment, a cluster, a compute pool, and API keys. - The Confluent CLI, to create a global API key and a connection for the AI agent in Step 6. For more information, see [Use API Keys to Authenticate to Confluent Cloud](../security/authenticate/workload-identities/service-accounts/api-keys/overview.md#cloud-api-keys). ## Step 1: Create an environment, a cluster, and a compute pool Create the resources in a region that supports managed AI models, Lightning Tables, and Tableflow. This quick start uses AWS `us-east-1`. For other regions, see [Confluent Cloud Regions and Availability by Cloud Provider](../get-started/regions.md#providers-regions). 1. In Confluent Cloud Console, create an environment named `river-air`. 2. In the environment, create a Standard cluster on AWS in `us-east-1`. 3. Create an Apache Flink® compute pool on AWS in `us-east-1`. 4. Open a SQL workspace on the compute pool, and set the catalog to `river-air` and the database to your cluster. For more information, see [Flink SQL Quick Start with Confluent Cloud Console](../flink/get-started/quick-start-cloud-console.md#flink-sql-quick-start-cloud-console). ## Step 2: Create the source tables River Air’s operational data arrives as three streams: flight status, passenger itineraries, and hotel room availability. Each table has a primary key and is backed by a compacted Apache Kafka® topic, so it holds the latest value for each key. 1. In the SQL workspace, run the following statements to create the tables. ```sql CREATE TABLE flight_status ( flight_id STRING, origin STRING, destination STRING, scheduled_time TIMESTAMP(3), estimated_time TIMESTAMP(3), status STRING, PRIMARY KEY (flight_id) NOT ENFORCED ) DISTRIBUTED BY (flight_id) INTO 1 BUCKETS WITH ('changelog.mode' = 'upsert', 'kafka.cleanup-policy' = 'compact'); CREATE TABLE passenger_itineraries ( passenger_id STRING, inbound_flight_id STRING, connecting_flight_id STRING, PRIMARY KEY (passenger_id) NOT ENFORCED ) DISTRIBUTED BY (passenger_id) INTO 1 BUCKETS WITH ('changelog.mode' = 'upsert', 'kafka.cleanup-policy' = 'compact'); CREATE TABLE hotel_inventory ( hotel STRING, available_rooms INT, nightly_rate DECIMAL(10, 2), PRIMARY KEY (hotel) NOT ENFORCED ) DISTRIBUTED BY (hotel) INTO 1 BUCKETS WITH ('changelog.mode' = 'upsert', 'kafka.cleanup-policy' = 'compact'); ``` 2. Run the following statements to add flights, three passengers on inbound flight RA417 from Chicago, and three airport hotels. ```sql INSERT INTO flight_status VALUES ('RA417', 'ORD', 'SFO', TIMESTAMP '2026-11-04 13:00:00', TIMESTAMP '2026-11-04 13:00:00', 'ON_TIME'), ('RA210', 'SFO', 'SEA', TIMESTAMP '2026-11-04 14:00:00', TIMESTAMP '2026-11-04 14:00:00', 'ON_TIME'), ('RA214', 'SFO', 'SEA', TIMESTAMP '2026-11-04 15:30:00', TIMESTAMP '2026-11-04 15:30:00', 'ON_TIME'), ('RA226', 'SFO', 'SEA', TIMESTAMP '2026-11-04 17:45:00', TIMESTAMP '2026-11-04 17:45:00', 'ON_TIME'), ('RA232', 'SFO', 'SEA', TIMESTAMP '2026-11-05 07:15:00', TIMESTAMP '2026-11-05 07:15:00', 'ON_TIME'); INSERT INTO passenger_itineraries VALUES ('P-001', 'RA417', 'RA210'), ('P-002', 'RA417', 'RA214'), ('P-003', 'RA417', CAST(NULL AS STRING)); INSERT INTO hotel_inventory VALUES ('Airport Bayfront Inn', 0, 189.00), ('Harbor Point Hotel', 12, 214.00), ('Runway View Suites', 5, 239.00); ``` ## Step 3: Keep one current state with a materialized table A materialized table keeps the result of a query continuously up to date. The following table joins each itinerary to its inbound and connecting flights and computes each passenger’s connection time and risk. It keeps one current row per passenger. 1. Run the following statement to create the materialized table. ```sql CREATE MATERIALIZED TABLE passenger_state ( PRIMARY KEY (passenger_id) NOT ENFORCED ) DISTRIBUTED BY (passenger_id) INTO 1 BUCKETS WITH ('changelog.mode' = 'upsert') AS SELECT p.passenger_id, p.inbound_flight_id, p.connecting_flight_id, COALESCE(onward.destination, inbound.destination) AS final_destination, inbound.estimated_time AS inbound_arrival, TIMESTAMPDIFF(MINUTE, inbound.estimated_time, onward.estimated_time) AS connection_minutes, CASE WHEN p.connecting_flight_id IS NULL THEN 'NO_CONNECTION' WHEN TIMESTAMPDIFF(MINUTE, inbound.estimated_time, onward.estimated_time) < 45 THEN 'AT_RISK' ELSE 'OK' END AS risk FROM passenger_itineraries p JOIN flight_status inbound ON p.inbound_flight_id = inbound.flight_id LEFT JOIN flight_status onward ON p.connecting_flight_id = onward.flight_id; ``` 2. Set the workspace **Mode** to **Snapshot**. 3. Run the following query. Results can take a minute or two to appear the first time, while the materialized table processes the source data. ```sql SELECT passenger_id, connecting_flight_id, inbound_arrival, connection_minutes, risk FROM passenger_state; ``` Your output should resemble: ```text passenger_id connecting_flight_id inbound_arrival connection_minutes risk P-001 RA210 2026-11-04 13:00 60 OK P-002 RA214 2026-11-04 13:00 150 OK P-003 NULL 2026-11-04 13:00 NULL NO_CONNECTION ``` For more information, see [Materialized Tables in Confluent Cloud for Apache Flink](../flink/concepts/materialized-tables.md#flink-sql-materialized-tables). ## Step 4: Delay the inbound flight Flight RA417 is now expected to land at 14:50, almost two hours late. Update its status, and watch the passenger state change. 1. Set the workspace **Mode** back to **Streaming**. 2. Run the following statement. ```sql INSERT INTO flight_status VALUES ('RA417', 'ORD', 'SFO', TIMESTAMP '2026-11-04 13:00:00', TIMESTAMP '2026-11-04 14:50:00', 'DELAYED'); ``` 3. In **Snapshot** mode, run the query from Step 3 again. Within about a minute, both connecting passengers are at risk. ```text passenger_id connecting_flight_id inbound_arrival connection_minutes risk P-001 RA210 2026-11-04 14:50 -50 AT_RISK P-002 RA214 2026-11-04 14:50 40 AT_RISK P-003 NULL 2026-11-04 14:50 NULL NO_CONNECTION ``` You didn’t change `passenger_state`. The materialized table recomputed the affected rows when the flight’s status changed. ## Step 5: Serve the current state with Lightning Tables A Lightning Table serves a topic’s current state for low-latency reads by applications and AI agents. 1. In Confluent Cloud Console, go to your cluster’s **Lightning Tables** page and click **Add Lightning Tables**. 2. For each of the `passenger_state`, `flight_status`, and `hotel_inventory` topics, click **Enable**, and then click **Enable** to confirm. Each table’s status changes from **Pending** to **Active** within about a minute. 3. To preview the data, click the code icon for `passenger_state` in the **Preview** column, and then click **Run** on the **SQL** tab. The results show the same three passengers as the snapshot query in Step 4. For more information, see [Enable the Real-Time Context Engine on a topic](../ai/real-time-context-engine/get-started.md#rtce-enable-topic) and [Lightning Tables in Confluent Cloud](../lightning-tables/overview.md#lightning-tables-overview). ## Step 6: Ground a rebooking agent in the current state A Streaming Agent can read Lightning Tables through the Real-Time Context Engine over the Model Context Protocol (MCP). In this step, you create an agent that runs on a managed model, looks up live flights and hotel rooms, and proposes two rebooking offers for each at-risk passenger. 1. Create a global API key for your user account. The agent’s MCP connection authenticates with it. ```bash confluent api-key create --resource global ``` Save the key and secret. Each user account can have only two global API keys. For more information, see [Use API Keys to Authenticate to Confluent Cloud](../security/authenticate/workload-identities/service-accounts/api-keys/overview.md#cloud-api-keys). 2. Create the connection to the Real-Time Context Engine MCP endpoint. Replace ``, ``, and `` with your IDs, and `` and `` with the global API key. If you created your resources in a region other than AWS `us-east-1`, also replace `aws` and `us-east-1` in the command and the endpoint URL with your cloud provider and region. ```bash confluent flink connection create rtce-connection \ --cloud aws --region us-east-1 --environment \ --type mcp_server \ --endpoint https://mcp.us-east-1.aws.confluent.cloud/mcp/v1/context-engine/organizations//environments//kafka-clusters/ \ --username --password \ --transport-type STREAMABLE_HTTP ``` Creating the connection with the CLI keeps the secret out of your SQL statement history. 3. In the SQL workspace, in **Streaming** mode, run the following statements to create a managed model, a tool that queries Lightning Tables over MCP, and the agent. The managed model runs in Confluent Cloud, so it needs no endpoint or API key of its own. ```sql CREATE MODEL rebooking_model INPUT (prompt STRING) OUTPUT (response STRING) WITH ( 'provider' = 'confluent', 'task' = 'text_generation', 'confluent.model' = 'anthropic.claude-sonnet-4-6' ); CREATE TOOL river_air_live_context USING CONNECTION `rtce-connection` WITH ( 'type' = 'mcp', 'allowed_tools' = 'getMetadata,queryData', 'request_timeout' = '30' ); CREATE AGENT rebooking_agent USING MODEL rebooking_model USING PROMPT 'You are the River Air rebooking agent. The passenger in the task missed a connection at SFO. Use the queryData tool (arguments: topic_name, query, max_result_rows) to read live data: 1. Query the flight_status topic for flights from SFO to the passenger''s final_destination. Offer the first two flights whose estimated_time is at least 45 minutes after the passenger''s inbound_arrival. Skip CANCELLED flights. 2. If an offered flight leaves 6 or more hours after inbound_arrival, query the hotel_inventory topic and choose the first hotel, in alphabetical order, with available_rooms above 0. Reply with exactly one line and nothing else: OFFER1=;OFFER2=;HOTEL=' USING TOOLS river_air_live_context WITH ('max_iterations' = '10'); ``` To see the managed models available in your region, run `SHOW MODELS`. 4. Agents read append-only input, so copy the at-risk passengers into a table of rebooking requests. Run the `CREATE TABLE` statement in **Streaming** mode. Then switch to **Snapshot** mode and run the `INSERT` statement. ```sql CREATE TABLE rebooking_requests ( passenger_id STRING, final_destination STRING, inbound_arrival TIMESTAMP(3) ); INSERT INTO rebooking_requests SELECT passenger_id, final_destination, inbound_arrival FROM passenger_state WHERE risk = 'AT_RISK'; ``` 5. In **Snapshot** mode, run the agent for each request. ```sql SELECT q.passenger_id, r.agent_status, r.response FROM rebooking_requests AS q, LATERAL TABLE(AI_RUN_AGENT( 'rebooking_agent', CONCAT('passenger_id=', q.passenger_id, '; final_destination=', q.final_destination, '; inbound_arrival=', CAST(q.inbound_arrival AS STRING)), q.passenger_id )) AS r(agent_status, response); ``` After about a minute, your output should resemble the following. The agent skips RA214, which now leaves only 40 minutes after landing, and adds a hotel for the overnight flight. The model might add a sentence before the `OFFER1` line, and its wording can vary. ```text passenger_id agent_status response P-001 SUCCESS OFFER1=RA226;OFFER2=RA232;HOTEL=Harbor Point Hotel P-002 SUCCESS OFFER1=RA226;OFFER2=RA232;HOTEL=Harbor Point Hotel ``` For more information, see [Streaming Agents with Confluent Intelligence in Confluent Cloud](../ai/streaming-agents/overview.md#ai-streaming-agents-overview), [Run Managed Model Inference with Confluent Cloud for Apache Flink](../ai/managed-model.md#flink-sql-managed-ai-model), [CREATE TOOL Statement in Confluent Cloud for Apache Flink](../flink/reference/statements/create-tool.md#flink-sql-create-tool), [CREATE AGENT Statement in Confluent Cloud](../flink/reference/statements/create-agent.md#flink-sql-create-agent), and [AI_RUN_AGENT](../flink/reference/functions/model-inference-functions.md#flink-sql-ai-run-agent-function). ## Step 7: Answer questions with history Tableflow materializes a topic as an Iceberg table, and a snapshot query reads the history in that table together with the newest events in Kafka. In this step, you record past flights, enable Tableflow on them, and ask how River Air’s flights have performed. Because you add these rows only minutes before you query them, the result can come entirely from Kafka. As the topic ages, the same query reads the older records from the Iceberg table. 1. In **Streaming** mode, run the following statements to create a flight history and add a few past flights. ```sql CREATE TABLE flight_history ( flight_id STRING, flight_date DATE, delay_minutes INT, affected_passengers INT ); INSERT INTO flight_history VALUES ('RA417', DATE '2026-09-24', 12, 0), ('RA417', DATE '2026-09-25', 0, 0), ('RA417', DATE '2026-10-02', 95, 120), ('RA210', DATE '2026-10-02', 40, 18), ('RA417', DATE '2026-10-09', 5, 0), ('RA214', DATE '2026-10-16', 130, 64); ``` 2. In Confluent Cloud Console, go to the `flight_history` topic and click **Enable Tableflow**. Keep **Iceberg** selected, and click **Use Confluent storage**. The topic shows **Tableflow sync pending** until the table is ready, in about a minute. 3. In **Snapshot** mode, run the following query. ```sql SELECT flight_id, COUNT(*) AS flights, SUM(CASE WHEN delay_minutes > 15 THEN 1 ELSE 0 END) AS delayed, SUM(affected_passengers) AS impacted FROM flight_history GROUP BY flight_id; ``` Your output should resemble: ```text flight_id flights delayed impacted RA214 1 1 64 RA210 1 1 18 RA417 4 1 120 ``` For more information, see [Tableflow in Confluent Cloud](../topics/tableflow/overview.md#cloud-tableflow) and [Snapshot Queries in Confluent Cloud for Apache Flink](../flink/concepts/snapshot-queries.md#flink-sql-snapshot-queries). ## Step 8: Clean up To stop incurring charges, delete the resources you created. 1. In Confluent Cloud Console, delete the `river-air` environment. Deleting the environment deletes its cluster, compute pool, topics, Lightning Tables, Tableflow tables, and Flink statements. 2. Delete the global API key that you created in Step 6: ```bash confluent api-key delete ``` ## Which Streamhouse solutions this quick start shows | Solution | Where it shows up in the quick start | Learn more | |-------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------| | Maintain one consistent state | `passenger_state` holds one current row per passenger, and the
preview panel and the agent read the same table. | [Maintain one consistent state across participants](capture.md#streamhouse-consistent-state) | | Decide across five seconds and five years | The agent acts on live state in Step 6, and a snapshot query in Step 7
answers questions about past flights from the Tableflow history. | [Decide across the last five seconds and the last five years](run.md#streamhouse-decide) | | Ground and adapt an AI agent | The rebooking agent reads live flights and hotel rooms through the
Real-Time Context Engine before it proposes offers. | [Ground and continuously adapt an AI agent](run.md#streamhouse-ground-ai) | | Monitor the business | `passenger_state` flags a connection as at risk when the time between
flights drops below 45 minutes. | [Monitor the business to prevent negative impact](run.md#streamhouse-monitor) | ## Next steps - For a fuller version of this scenario, with an operations app, a live data stream, and 30 days of history, deploy the [quickstart-streamhouse repository](https://github.com/confluentinc/quickstart-streamhouse) to your own Confluent Cloud environment. - For the pattern behind this quick start, see [Start here: the always-current table](query-surface.md#streamhouse-query-start-here). - To go further, try two tutorials on Confluent Developer, each in three parts: [Streamhouse for Analytics](https://developer.confluent.io/confluent-tutorials/analytics-tableflow/), which adds Spark and a live dashboard, and [Streamhouse for AI Agents](https://developer.confluent.io/confluent-tutorials/agentic-ai-model-tool-setup/flinksql/), which builds a Streaming Agent that calls a tool over MCP. ## Related content - [Build a Streamhouse on Confluent Cloud](overview.md#streamhouse-overview) - [Choose a query surface](query-surface.md#streamhouse-query-choose-surface) - [Flink SQL Quick Start with Confluent Cloud Console](../flink/get-started/quick-start-cloud-console.md#flink-sql-quick-start-cloud-console) #### 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).