Loading with Snowpipe and similar tools

Loading with Snowpipe and similar tools

Tools such as Snowpipe and Databricks Auto Loader load new files as they appear, and BigQuery Data Transfer Service loads them on a schedule. They only ever append rows. They don’t know about markers, deleted rows or snapshots, and don’t promise to load files in order. Pointed straight at the bucket, they keep every version of a row, never remove deleted rows, and pile each snapshot on top of the last one.

They work well as the first step, though. Load every file into a raw table, with its file name. Then work out your current rows from the raw table:

  1. Only use complete markers, in order. A marker is complete once every file it lists is loaded, with as many rows as the marker says. Files with no rows don’t need loading. Stop at the first marker that isn’t complete: leave it and every marker after it out until it is.
  2. Start from each table’s latest snapshot. For each table, only use the newest complete snapshot marker that lists it and the markers after it.
  3. Take each row’s latest version. For each _row_id, take the row from the marker with the latest name, and within a marker from the file listed last. Leave the row out if _deleted is true there.

This gives the same result as loading markers one at a time in order, whatever order the files load in. While files are still arriving, a table only has rows once its first snapshot has loaded. A marker can only hold back later ones once its own file has loaded, so until then you can briefly see a later batch’s changes without an earlier one’s.

Snowflake example

This loads every file into two raw tables with Snowpipe, then works out your current rows in a view. It assumes your files are under rollerup/ in your-bucket. Create the stage at the bucket root, so that each file’s name matches the paths in the markers.

CREATE STAGE rollerup_stage
  URL = 's3://your-bucket/'
  STORAGE_INTEGRATION = your_integration;

CREATE TABLE rollerup_raw_rows (
  file STRING,
  data VARIANT,
  loaded_at TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE TABLE rollerup_raw_markers (
  file STRING,
  data VARIANT,
  loaded_at TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE PIPE rollerup_rows_pipe AUTO_INGEST = TRUE AS
  COPY INTO rollerup_raw_rows (file, data)
  FROM (SELECT METADATA$FILENAME, $1 FROM @rollerup_stage)
  PATTERN = 'rollerup/.*[.]parquet'
  FILE_FORMAT = (TYPE = PARQUET);

CREATE PIPE rollerup_markers_pipe AUTO_INGEST = TRUE AS
  COPY INTO rollerup_raw_markers (file, data)
  FROM (SELECT METADATA$FILENAME, $1 FROM @rollerup_stage)
  PATTERN = 'rollerup/_batches/.*[.]json'
  FILE_FORMAT = (TYPE = JSON);

AUTO_INGEST needs an event notification on your bucket; see Snowflake’s Snowpipe docs. Each Parquet row loads as one object, so new columns come through without changing anything.

Snowpipe only loads files that arrive after you create a pipe. If the data feed was already running, create the pipes first, then load the files already there by running each pipe’s COPY INTO once by itself. Loading a file twice is fine: the views below ignore repeats.

The complete markers, up to the first one that isn’t, one row per file they list:

CREATE VIEW rollerup_markers AS
WITH markers AS (
  SELECT
    REGEXP_REPLACE(file, '^.*/|[.]json$', '') AS name,
    data:kind::STRING AS kind,
    data:files AS files
  FROM rollerup_raw_markers
  QUALIFY ROW_NUMBER() OVER (PARTITION BY file ORDER BY loaded_at DESC) = 1
),
listed AS (
  SELECT
    m.name,
    m.kind,
    f.index AS position,
    f.value:table::STRING AS table_name,
    f.value:path::STRING AS path,
    f.value:rows::INTEGER AS row_count
  FROM markers m, LATERAL FLATTEN(input => m.files) f
),
loaded AS (
  SELECT file, COUNT(DISTINCT data:_row_id::STRING) AS row_count
  FROM rollerup_raw_rows
  GROUP BY file
),
incomplete AS (
  SELECT DISTINCT l.name
  FROM listed l
  LEFT JOIN loaded r ON r.file = l.path
  WHERE l.row_count > 0 AND COALESCE(r.row_count, 0) < l.row_count
)
SELECT *
FROM listed l
WHERE NOT EXISTS (SELECT 1 FROM incomplete i WHERE i.name <= l.name);

Every table’s current rows:

CREATE VIEW rollerup_current_rows AS
WITH latest_snapshot AS (
  SELECT table_name, MAX(name) AS name
  FROM rollerup_markers
  WHERE kind = 'snapshot'
  GROUP BY table_name
)
SELECT m.table_name, r.data:_row_id::STRING AS row_id, r.data
FROM rollerup_markers m
JOIN latest_snapshot s ON s.table_name = m.table_name AND m.name >= s.name
JOIN rollerup_raw_rows r ON r.file = m.path
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY m.table_name, r.data:_row_id::STRING
    ORDER BY m.name DESC, m.position DESC
  ) = 1
  AND NOT r.data:_deleted::BOOLEAN;

Then a view per table, with the columns you need from Tables:

CREATE VIEW visits AS
SELECT
  data:id::STRING AS id,
  data:status::STRING AS status,
  data:locationId::STRING AS location_id,
  data:createdAt::TIMESTAMP_NTZ AS created_at
FROM rollerup_current_rows
WHERE table_name = 'visits';

As the raw tables grow, make rollerup_current_rows a dynamic table, or delete raw rows from markers older than their table’s latest snapshot.

Other tools

The same steps work elsewhere. With Databricks Auto Loader, add the file’s path from _metadata.file_path to each row; with BigQuery or Redshift, load into a raw table the same way.

Previous Snapshots Next Tables