4. Build the pipeline

Avro schemas, a PostgreSQL CDC source, two raw topics, a Flink join to a canonical topic, and two data products.

Part 4 of 10 in the Hello World tutorial · Follow this part in the guided wizard · Printable version

Your browser blocked the wizard window. How to allow it, or open the wizard in a new tab.

This is the longest part: you build the whole pipeline. In order:

1. rename the pipeline;
2. create five Avro schemas;
3. add the CDC source and its two raw topics;
4. add the Flink join and the canonical topic;
5. add the two data product statements and their topics.

4.1 Rename the pipeline

1. In the explorer, right-click data-product.kpd under Pipelines and choose Rename…

2. Type store-orders and click Rename. The file becomes store-orders.kpd.

4.2 Create the five Avro schemas

Every topic carries messages of a known shape, described by a schema. KPD has a visual editor for Avro, so you never need to write the JSON by hand. You need five: one for each raw table, one for the canonical order, and one for each data product. Their file names follow the topic names with -value on the end (the Schema Registry subject for a topic's values).

1. In the explorer, scroll down to Shared schemas, right-click it and choose New schema…

2. The New schema dialog opens with an example record. Keep Folder as Shared schemas (top level) and Format as Avro (.avsc).

3. File name: hwc.prod.raw.order.shopdb.v1-value
4. Record name: Order. Namespace: com.hellocoffee.raw
5. Change the two example fields and use Add field until you have these five. Type each name, then pick its type from the list (timestamp-millis is in the list too):

   Field	Type
   order_id	long
   customer_id	long
   store_id	string
   total_pence	int
   ordered_at	timestamp-millis

Two handy extras while you are here. Open Example message to see a sample record that matches the schema. Click Source to see the Avro JSON KPD writes for you. The blue Valid Avro bar means the schema is checked against the Avro specification as you type.

6. Click Create schema.
7. Repeat for the other four schemas:

   File name	Record / namespace	Fields
   hwc.prod.raw.customer.shopdb.v1-value	Customer / com.hellocoffee.raw	customer_id long, name string, email string (tick nullable)
   hwc.prod.ent.order.cdm.v1-value	StoreOrder / com.hellocoffee.cdm	orderId long, customerId long, customerName string, customerEmail string (nullable), storeId string, totalPence int, orderedAt timestamp-millis
   hwc.prod.ent.order.history.v1-value	OrderHistory / com.hellocoffee.product	orderId long, customerId long, customerName string, storeId string, totalPence int, orderedAt timestamp-millis
   hwc.prod.ent.order.big-orders.v1-value	BigOrder / com.hellocoffee.product	orderId long, storeId string, totalPence int, orderedAt timestamp-millis

4.3 Add the CDC source

CDC (change data capture) reads the database's own change log, so every insert and update is captured once, in order, without querying the tables.

1. Make sure the store-orders.kpd tab is showing. On the Start your pipeline card, click Add a source. (Start from a system would draw the shop database itself first, which is a good way into a workshop; here we go straight to the connector.)
2. Type postgres and choose Postgres CDC (its full name is PostgreSQL CDC Source V2).

3. The connector appears on the canvas. If it isn't inside the green Ingestion box, drag it there: sources belong in the ingestion layer.
4. On its General tab: Name hwc-shopdb-cdc; Description Captures every new and changed row in the shop database's orders and customers tables, one raw topic per table. (Leave Service account for Part 6.)

5. Connector tab: database.hostname shopdb.hello-coffee.example, database.port 5432, database.dbname shopdb, database.sslmode require. (No password: that is a secret, supplied when deploying.)

6. Capture tab: table.include.list public.orders,public.customers; snapshot.mode initial (copy the existing rows first, then stream changes); slot.name hwc_shopdb_slot; publication.name hwc_shopdb_pub; publication.autocreate.mode filtered (the publication covers just these two tables).

7. Data format tab: output.data.format AVRO, output.key.format STRING, topic.prefix hwc.prod.raw.shopdb.

8. Advanced tab: in SMT chain, type RegexRouter and press Enter. RegexRouter is the single message transform that renames each topic to our DPCF name.

Because a transform renames the topics, KPD can't predict the names any more, so it checks only that there is one topic per captured table. (The transform's own settings, the pattern and the replacement, are added to the Terraform by hand.)

4.4 Add the two raw topics

1. With the connector selected, click the + on its right-hand edge (its Records output).

2. Choose Topic. A topic appears, already connected.

3. On the topic's General tab: Name hwc.prod.raw.order.shopdb.v1; Description Every order row from shopdb (public.orders), as captured. Raw: never published.
4. Partitions & durability tab: Partitions 3; Partition key order_id (a design note: which field keeps one order's changes in order).

5. Schema tab: Key format String; Value schema file hwc.prod.raw.order.shopdb.v1-value.avsc.

6. Select the connector again, click its + again and add a second Topic: name hwc.prod.raw.customer.shopdb.v1, description Every customer row from shopdb (public.customers), as captured. Raw: never published., 3 partitions, partition key customer_id, key format String, value schema hwc.prod.raw.customer.shopdb.v1-value.avsc.

4.5 Join orders to customers in Flink

1. Select the order raw topic and click its +. Choose Flink SQL. The statement is placed in the purple Process box.

2. On the statement's General tab: Name orders_raw_to_cdm; Description Joins each raw order with its customer to make the canonical store order: business field names, the customer's name and email added.; Compute pool hwc-raw-to-cdm.

3. Now wire the customer topic in as the join. A Flink statement has a Main input on its left and a Join input on its bottom edge. Drag from the customer topic's right-hand dot (its output) to the dot on the bottom of the statement. A line follows the mouse while you drag.

4. SQL tab: Statement kind INSERT INTO. Keep the SQL in a file of its own, so it can be reviewed like any other code: under SQL file choose ＋ New SQL file…. The dialog suggests the name orders_raw_to_cdm in a new store-orders folder; keep both, replace the starter text with this, and click Create SQL file:

   -- Each order, with the customer who placed it.
   INSERT INTO `hwc.prod.ent.order.cdm.v1`
   SELECT
     o.order_id    AS orderId,
     o.customer_id AS customerId,
     c.name        AS customerName,
     c.email       AS customerEmail,
     o.store_id    AS storeId,
     o.total_pence AS totalPence,
     o.ordered_at  AS orderedAt
   FROM `hwc.prod.raw.order.shopdb.v1` AS o
   JOIN `hwc.prod.raw.customer.shopdb.v1` AS c
     ON o.customer_id = c.customer_id;

   The SQL tab now shows the file, Read-only so it isn’t changed by accident. To change it later, untick Read-only and edit it there, or open it from Stream configuration → SQL in the explorer.

4.6 Add the canonical topic

1. Select orders_raw_to_cdm, click the + on its Output and choose Topic.
2. Name hwc.prod.ent.order.cdm.v1; Description Canonical store orders: every order with its customer, in business field names. The single source for the data products.; 3 partitions, partition key orderId; key format String; value schema hwc.prod.ent.order.cdm.v1-value.avsc.

4.7 Add the two data products

Each data product is a Flink statement that reads the canonical topic and writes its own topic, in the blue Surface box.

1. Select the canonical topic, click its + and choose Flink SQL. Name it order_history_product, description Data product DP-001: every order with who placed it, for the app's order history., compute pool hwc-cdm-to-surface. On SQL, choose ＋ New SQL file… as before, keep the suggested name, and create it with:

   -- DP-001: each customer's orders, without their email.
   INSERT INTO `hwc.prod.ent.order.history.v1`
   SELECT orderId, customerId, customerName, storeId, totalPence, orderedAt
   FROM `hwc.prod.ent.order.cdm.v1`;

2. Select order_history_product, click the + on its Output, choose Topic: name hwc.prod.ent.order.history.v1, description DP-001 Customer order history: every store order with the customer who placed it., 3 partitions keyed by orderId, key String, value schema hwc.prod.ent.order.history.v1-value.avsc.
3. Select the canonical topic again and add a second Flink SQL: big_orders_product, description Data product DP-002: orders of £20 or more, for store managers., pool hwc-cdm-to-surface, and a ＋ New SQL file… with:

   -- DP-002: orders of £20 (2000 pence) or more.
   INSERT INTO `hwc.prod.ent.order.big-orders.v1`
   SELECT orderId, storeId, totalPence, orderedAt
   FROM `hwc.prod.ent.order.cdm.v1`
   WHERE totalPence >= 2000;

4. Give it an output Topic: hwc.prod.ent.order.big-orders.v1, description DP-002 Big orders: orders of £20 or more, as they happen., 3 partitions keyed by orderId, key String, value schema hwc.prod.ent.order.big-orders.v1-value.avsc.

Save. Next: 5. Governance: tags and the catalogue.