Scenario F: Privacy-safe ingestion¶

Part of Streaming Transformations › Scenarios.

Goal¶

Personal data such as email addresses must not land in clear text, and records from users who did not give consent must not land at all.

Transformations used¶

Custom scalar logic: a SQL UDF replaces personal identifiers with salted hashes in the SELECT list.
Stream filtering on stream columns: ingest only records with a consent flag.
Stream enrichment: attach audience segments.

Source data¶

Stream¶

Each streamed record looks like this:

{"ts": "2026-10-09 10:00:00", "customer_id": "c-1001", "email": "Ana.Lopez@example.com", "phone": "+1-555-010-4821", "country": "US", "consent": true}
{"ts": "2026-10-09 10:00:05", "customer_id": "c-1002", "email": "jan.novak@example.org", "phone": "+49-30-5550-7713", "country": "DE", "consent": true}
{"ts": "2026-10-09 10:00:10", "customer_id": "c-1003", "email": "Mei.Chen@example.net", "phone": "+81-3-5550-2290", "country": "JP", "consent": false}

Dimension table¶

The PIPE joins this dimension table, refreshed periodically:

-- Audience segment dimension: marketing segment per customer.
CREATE OR REPLACE TABLE dim_audience_segments (
  customer_id STRING PRIMARY KEY RELY,
  audience_segment STRING);

INSERT INTO dim_audience_segments VALUES
  ('c-1001', 'frequent_buyer'),
  ('c-1002', 'new_signup');

Target table¶

The PIPE lands redacted and enriched records of consenting users in this target table:

CREATE OR REPLACE TABLE customer_signups (
  event_ts TIMESTAMP_NTZ, customer_id STRING, email_hash STRING,
  phone_masked STRING, country STRING, audience_segment STRING);

Transformation¶

The PIPE lands records in the target table. Records of users who did not consent are filtered out. Email addresses are replaced with salted hashes, and phone numbers are masked. Audience segments are attached by joining the dimension table.

-- SQL UDF: salted SHA-256 of the lowercased value, so the raw identifier never lands.
CREATE OR REPLACE FUNCTION hash_pii(v STRING) RETURNS STRING AS
  $$ SHA2(CONCAT('st-demo-salt:', LOWER(v)), 256) $$;

-- Pipe with enrichment and filter, calling UDF in projection
CREATE OR REPLACE PIPE customer_signups_pipe AS
  COPY INTO customer_signups (event_ts, customer_id, email_hash, phone_masked,
                              country, audience_segment) FROM (
    SELECT
      e.$1:ts::TIMESTAMP_NTZ,
      e.$1:customer_id::STRING,
      hash_pii(e.$1:email::STRING),
      CONCAT('***-***-', RIGHT(e.$1:phone::STRING, 4)),
      e.$1:country::STRING,
      s.audience_segment
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) e
    LEFT OUTER JOIN dim_audience_segments s
      ON e.$1:customer_id::STRING = s.customer_id
    WHERE e.$1:consent::BOOLEAN = TRUE);

Result¶

Querying the target table returns rows like these:

EVENT_TS	CUSTOMER_ID	EMAIL_HASH	PHONE_MASKED	COUNTRY	AUDIENCE_SEGMENT
2026-10-09 10:00:00	c-1001	788672793c07c4e4b1e19cb82497fe54ed9832ab0553fa51e10cd93551e3c9c9	*-*-4821	US	frequent_buyer
2026-10-09 10:00:05	c-1002	97adc09b6694caeb9cadcab275c61b0472d722c58b6c4ac5bf56b40fe7b7b24f	*-*-7713	DE	new_signup

The record for c-1003 is dropped because its consent flag is false. The raw email address and phone number never reach the table, and the email is lowercased before salting and hashing, so the same address always produces the same hash regardless of capitalization.

Considerations¶

Dimension data is a snapshot that refreshes periodically, so updates can take several minutes to appear. This works well for slow-moving reference data. Dynamic tables can be used as dimension tables.