# Stream Processing with ksqlDB for Confluent Platform
Stream processing is a form of computing that runs continually over
unbounded streams of events, transforming, filtering, joining, or
aggregating them until you explicitly stop the process. In
ksqlDB, you do this by deriving new collections from existing ones:
when a collection is updated with a new event, ksqlDB updates the
collections derived from it in real time.
Declaring a collection with an enforced schema over a new or existing
Apache Kafka® topic is useful on its own, but only for working with events in
their current form—stream processing is what lets you transform,
filter, join, and aggregate them into an application.
The general pattern for stream processing in ksqlDB is to create a new
collection by using the `SELECT` statement on an existing collection.
The result of the inner `SELECT` feeds into the outer declared
collection. You don’t need to declare a schema when deriving a new
collection, because ksqlDB infers the column names and types from the
inner `SELECT` statement. The value of the `ROWTIME` pseudo column
defines the timestamp of the record written to Kafka, and the
value of the `ROWPARTITION` and `ROWOFFSET` pseudo columns define
the partition and offset of the source record, respectively. The value
of system columns can not be set in the `SELECT`.
Here are a few examples of deriving between the different collection
types.
## Derive a new stream from an existing stream
Given the following stream:
```sql
CREATE STREAM rock_songs (artist VARCHAR, title VARCHAR)
WITH (kafka_topic='rock_songs', partitions=2, value_format='avro');
```
You can derive a new stream with all of the song titles transformed to
uppercase:
```sql
CREATE STREAM title_cased_songs AS
SELECT artist, UCASE(title) AS capitalized_song
FROM rock_songs
EMIT CHANGES;
```
Each time a new song is inserted into the `rock_songs` topic, the
uppercase version of the title is appended to the `title_cased_songs`
stream.
## Deriving a new table from an existing stream
Given the following table and stream:
```sql
CREATE TABLE products (product_name VARCHAR PRIMARY KEY, cost DOUBLE)
WITH (kafka_topic='products', partitions=1, value_format='json');
CREATE STREAM orders (product_name VARCHAR KEY)
WITH (kafka_topic='orders', partitions=1, value_format='json');
```
You can create a table that aggregates rows from the `orders` stream,
while also joining the stream on the `products` table to enrich the
`orders` data:
```sql
CREATE TABLE order_metrics AS
SELECT p.product_name, COUNT(*) AS count, SUM(p.cost) AS revenue
FROM orders o JOIN products p ON p.product_name = o.product_name
GROUP BY p.product_name EMIT CHANGES;
```
This aggregate table keeps track of the total number of orders of each
product, along with the total amount of revenue generated by each
product.
## Deriving a new table from an existing table
Given the following aggregate table:
```sql
CREATE TABLE page_view_metrics AS
SELECT url, location_id, COUNT(*) AS count
FROM page_views GROUP BY url EMIT CHANGES;
```
You can derive another table that filters out rows from the
`page_view_metrics` table:
```sql
CREATE TABLE page_view_metrics_mountain_view AS
SELECT url, count FROM page_view_metrics
WHERE location_id = 42 EMIT CHANGES;
```
## Deriving a new stream from multiple streams
Given the following two streams:
```sql
CREATE STREAM impressions (user VARCHAR KEY, impression_id BIGINT, url VARCHAR)
WITH (kafka_topic='impressions', partitions=1, value_format='json');
CREATE STREAM clicks (user VARCHAR KEY, url VARCHAR)
WITH (kafka_topic='clicks', partitions=1, value_format='json');
```
You can create a derived stream that joins the `impressions` and
`clicks` streams to output rows indicating that a given impression has
been clicked within one minute of the initial ad impression:
```sql
CREATE STREAM clicked_impressions AS
SELECT * FROM impressions i JOIN clicks c WITHIN 1 minute ON i.user = c.user
WHERE i.url = c.url
EMIT CHANGES;
```
Any time an `impressions` row is received, followed within one minute
by a `clicks` row having the same `user`, a row is emitted into the
`clicked_impressions` stream.