5 种流式 ETL 模式

-- Filter only records that modify the inventoryinsert into inventory_updatesselect * from http_eventswhere hostname = 'api.mycompany.com' andpath like '/v1/inventory%' and
  method in ( 'POST', 'PUT', 'DELETE', 'PATCH' )

-- Route security-related HTTP eventsinsert into security_eventsselect * from http_eventswhere path like '/login%' orpath like '/billing/cc%'

-- Route requests to Customer Success if it looks like the user needs helpinsert into cs_alertsselect * from http_eventswhere response_code between 500 and 599 or -- any server failure( path = '/signup' and response_code != 200 ) or -- failed to sign up for any reason

-- Parse timestamp and actioninsert into user_eventsselectto_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', 'PUT' then 'modify'
    when 'DELETE'      then 'delete'
  end as actionfrom (  select
    grok(
      body,      '\[${ISO8661_DATETIME:ts} ${DATA:method} "${PATH:path}" uid:${DATA:user_id}'
    ) as fields  from http_event
)

-- Cleanse incoming data for downstream processesinsert into sensor_readingsselectcast(ifnull(sensor_id, '0') as bigint) as sensor_id,  lower(trim(name))                      as name,  cast(`value` as bigint)                as readingfrom raw_sensor_readings

-- Anonymize SSNs and zip codesinsert into user_events_maskedselectuser_id,
  username,  overlay(ssn placing '*' from 1 for 12) as ssn,  substring(zip_code from 1 for 2)       as zip_code_1,
actionfrom user_events

-- Count the number of events by path and status every 10 seconds.insert into site_activityselectwindow_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

-- Build hourly usage data for a Stripe integration on the output streaminsert into stripe_product_usageselectwindow_start as _time,
  customer_id,  'abcd1234' as price_id  sum(bytes_sent) / 1024 / 1024 as mb_sentfrom table(
  tumble(    table document_downloads,    descriptor(_time),
    interval '1' hour
  )
)group by window_start, customer_idhaving mb_sent > 1024