5 Streaming ETL Patterns Using SQL

Using SQL for Transformations?

SQL combines declarative power with a skill set that almost every developer or analyst possesses, making it a natural fit for implementing ETL/ELT transformations.

Pipeline Patterns

Most ETL pipelines conform to one or more of the patterns below. Decodable’s stream‑pipeline abstraction lets you build a single pipeline or decompose complex transformations into reusable, connected pipelines.

1. Filter

Filters discard rows that do not satisfy a WHERE clause. Typical uses include enforcing compliance, reducing processing load, or lowering storage requirements.

-- Filter only records pertaining to the application
insert into application_events
select *
from http_events
where hostname = 'app.decodable.co';

-- Filter only records that modify the inventory
insert into inventory_updates
select *
from http_events
where hostname = 'api.mycompany.com'
  and path like '/v1/inventory%'
  and method in ('POST','PUT','DELETE','PATCH');

2. Route

The Route pattern creates multiple output streams from one or more input streams, directing records to the appropriate destination based on a set of rules. It is effectively a collection of filters, each passing records that match its specific criteria.

-- Route security‑related HTTP events
insert into security_events
select *
from http_events
where path like '/login%'
   or path like '/billing/cc%';

-- Route app‑related HTTP events
insert into application_events
select *
from http_events
where hostname = 'app.decodable.co';

-- Route alerts for server failures or signup problems
insert into cs_alerts
select *
from http_events
where response_code between 500 and 599
   or (path = '/signup' and response_code != 200);

3. Transform

Transformation pipelines modify input records to produce output records. Most transformations are 1:1, but some produce one‑to‑many relationships.

Transform: Extract

Parsing input records extracts data that becomes the basis for enriched output records.

-- Parse timestamp and action
insert into user_events
select
  to_date(fields['ts'], 'YYYY-MM-DD''T''HH:MI:SS') as ts,
  fields['user_id'] as user_id,
  fields['path'] as path,
  case fields['method']
    when 'GET'  then 'read'
    when 'POST' then 'modify'
    when 'PUT'  then 'modify'
    when 'DELETE' then 'delete'
  end as action
from (
  select grok(body, '[${ISO8661_DATETIME:ts} ${DATA:method} "${PATH:path}" uid:${DATA:user_id}') as fields
  from http_event
);

Transform: Normalize

Incoming records often need to be normalized to a schema that downstream systems can handle, filling missing fields, dropping optional ones, and enforcing data types.

-- Cleanse incoming data for downstream processes
insert into sensor_readings
select
  cast(ifnull(sensor_id, '0') as bigint) as sensor_id,
  lower(trim(name)) as name,
  cast(`value` as bigint) as reading
from raw_sensor_readings;

Transform: Anonymize

When the target system does not need certain information, an anonymization pipeline removes or masks sensitive fields for compliance or privacy reasons.

-- Anonymize SSNs and zip codes
insert into user_events_masked
select
  user_id,
  username,
  overlay(ssn placing '*' from 1 for 12) as ssn,
  substring(zip_code from 1 for 2) as zip_code_1,
  action
from user_events;

4. Aggregate

Aggregation pipelines typically use SQL window functions to bucket incoming records (often by time) and then apply aggregate operators such as COUNT, MIN, MAX, AVG, SUM.

-- Count the number of events by path and status every 10 seconds.
insert into site_activity
select
  window_start,
  window_end,
  path,
  status,
  count(1) as `count`
from table(
  tumble(table http_events, descriptor(_time), interval '10' seconds)
)
group by window_start, window_end, path, status;

5. Trigger

The Trigger pattern differs from the others: its output records may have little overlap with the input schema because they represent alerts generated when a set of conditions is detected across one or more input records.

-- Build hourly usage data for a Stripe integration on the output stream
insert into stripe_product_usage
select
  window_start as _time,
  customer_id,
  'abcd1234' as price_id,
  sum(bytes_sent) / 1024 / 1024 as mb_sent
from table(
  tumble(table document_downloads, descriptor(_time), interval '1' hour)
)
group by window_start, customer_id
having mb_sent > 1024;