How To Impelement PII Masking Apache Iceberg Using RisingWave

In this blog, i will share how to implement PII Masking for sensitive data like address, name, email from Postgresql into Apache Iceberg using RisingWave as streaming tools.

Why PII Masking Matters for Data Security?

In today’s data-driven world, organizations are collecting more personal data than ever before. Personally Identifiable Information (PII) — such as names, email addresses, phone numbers, and national IDs — is often stored across various databases, lakes, and warehouses to power analytics and AI. However, with this growing dependency on data comes an equally growing responsibility: keeping sensitive information safe.

PII masking is a critical technique for ensuring that sensitive user information is not exposed to unauthorized access or misuse. Instead of storing raw data, masking transforms PII into obfuscated or anonymized formats while maintaining data utility for downstream processes.

Here are key reasons why PII masking is essential:

• Regulatory Compliance: Regulations like GDPR, HIPAA, and CCPA mandate strict data privacy controls. Failure to properly protect PII can lead to heavy fines and legal consequences.
• Minimized Data Breach Risk: Masking reduces the potential damage from data breaches. Even if attackers gain access to the data, the masked PII is far less useful.
• Safe Data Sharing: Teams often need to share datasets with analysts, developers, or third parties. Masking ensures that data can be shared without exposing sensitive personal details.
• Preserving Trust: Protecting customer data is not just about compliance — it’s about maintaining trust. Users are more likely to engage with businesses that prioritize their privacy.

When building a modern data pipeline — such as streaming data from PostgreSQL to an Iceberg data lake using RisingWave — integrating PII masking early in the process ensures that your architecture is secure by design.

Let’s implement it using RisingWave

For the first time, we will use several tools such as minio, postgresql, risingwave which are available in docker compose on my github.

https://github.com/dwickyfp/implement-pii-mask-iceberg-using-risingwave

Next, we will create dummy data in postgres

CREATE TABLE users_sensitive_data (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255),
    phone VARCHAR(50),
    full_name VARCHAR(255),
    address TEXT,
    passport_number VARCHAR(50),
    national_id VARCHAR(50),
    credit_card_number VARCHAR(25),
    ssn VARCHAR(20),
    date_of_birth DATE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO users_sensitive_data (
    email, phone, full_name, address, passport_number,
    national_id, credit_card_number, ssn, date_of_birth
) VALUES
('alice.smith@example.com', '+1-202-555-0143', 'Alice Smith', '123 Apple St, New York, USA', 'A1234567', '987654321', '4111111111111111', '123-45-6789', '1985-06-15'),
('bob.johnson@example.com', '+44 20 7946 0958', 'Bob Johnson', '45 Baker St, London, UK', 'B7654321', '876543219', '5500000000000004', '987-65-4321', '1990-02-20'),
('charlie.lee@example.com', '+62 812-3456-7890', 'Charlie Lee', 'Jl. Merdeka No. 1, Jakarta, Indonesia', 'C2345678', '765432198', '340000000000009', '111-22-3333', '1988-11-11'),
('danielle.wang@example.com', '+86 10 8888 8888', 'Danielle Wang', '88 Nanjing Rd, Shanghai, China', 'D8765432', '654321987', '6011000000000004', '444-55-6666', '1995-07-07'),
('emmanuel.nguyen@example.com', '+33 1 44 55 66 77', 'Emmanuel Nguyen', '22 Rue de Lyon, Paris, France', 'E3456789', '543219876', '3530111333300000', '777-88-9999', '1982-01-01'),
('fatima.khan@example.com', '+971 50 123 4567', 'Fatima Khan', '101 Palm St, Dubai, UAE', 'F6543210', '432198765', '4111111111111111', '222-33-4444', '1993-03-12'),
('george.rodriguez@example.com', '+34 91 123 4567', 'George Rodriguez', '5 Gran Via, Madrid, Spain', 'G0987654', '321987654', '4000000000000002', '555-66-7777', '1987-08-22'),
('hannah.chen@example.com', '+81 3-1234-5678', 'Hannah Chen', '9 Shibuya, Tokyo, Japan', 'H4567890', '210987543', '6011000990139424', '999-00-1111', '1991-09-09'),
('ivan.petrov@example.com', '+7 495 123-45-67', 'Ivan Petrov', '15 Tverskaya St, Moscow, Russia', 'I5678901', '109876432', '378282246310005', '333-44-5555', '1980-04-04'),
('julia.almeida@example.com', '+55 11 91234-5678', 'Julia Almeida', 'Rua Augusta, São Paulo, Brazil', 'J6789012', '098765321', '6011000000000004', '666-77-8888', '1989-12-25');

After that, we will create sink data from postgres into RisingWave using postgres-cdc, for complete script :

CREATE SOURCE pg_source WITH (
    connector='postgres-cdc',
    hostname='postgres-vendor-0',
    port='5432',
    username='postgres',
    password='postgres',
    database.name='postgres',
    schema.name='public',
    slot.name = 'rising_wave',
    publication.name ='rw_publication'
);

CREATE TABLE users_sensitive_data (
    id INT PRIMARY KEY,
    email VARCHAR,
    phone VARCHAR,
    full_name VARCHAR,
    address VARCHAR,
    passport_number VARCHAR,
    national_id VARCHAR,
    credit_card_number VARCHAR,
    ssn VARCHAR,
    date_of_birth DATE,
    created_at TIMESTAMP 
) FROM pg_source TABLE 'public.users_sensitive_data';

Next, let’s make udf python as external function in RisingWave, for documentation, you check in :

[embed]External Python UDFs - RisingWave This article provides a step-by-step guide for defining and running external Python UDFs, and calling them from…docs.risingwave.com

I have created a sample implementation of a UDF external function, which is available in my GitHub repository :

https://github.com/dwickyfp/implement-pii-mask-iceberg-using-risingwave

Please install requirements.txt before running this python file, after that run “python udf.py”.

Next, in RisingWave create external function :

CREATE FUNCTION mask_pii(VARCHAR, VARCHAR) RETURNS VARCHAR
AS mask_pii USING LINK 'http://<net-host>:8815';

Than you can using external function in select statement like this :

select 
   id,
   mask_pii(email, 'email') as "email",
   mask_pii(phone, 'phone') as "phone",
   mask_pii(full_name, 'name') as "full_name",
   mask_pii(address, 'address') as "address",
   mask_pii(passport_number, 'passport') as "passport_number",
   mask_pii(national_id, 'national_id') as "national_id",
   mask_pii(credit_card_number, 'credit_card') as "credit_card_number",
   mask_pii(ssn, 'ssn') as "ssn",
   mask_pii(cast(date_of_birth as VARCHAR), 'dob') as "date_of_birth"
from users_sensitive_data

Based on this data, create materialized view than sink into Apache Iceberg and usually i use Apache Amoro as a Rest Catalog. You can check how to setup Apache Amoro as Rest Catalog in this link :

[embed]Quickstart This guide outlines the basic process of using Amoro, allowing you to quickly experience its core features. You can…amoro.apache.org

Than create sink into Apache Iceberg

create materialized view users_sensitive_data_masking as 
select 
 id,
 mask_pii(email, 'email') as "email",
 mask_pii(phone, 'phone') as "phone",
 mask_pii(full_name, 'name') as "full_name",
 mask_pii(address, 'address') as "address",
 mask_pii(passport_number, 'passport') as "passport_number",
 mask_pii(national_id, 'national_id') as "national_id",
 mask_pii(credit_card_number, 'credit_card') as "credit_card_number",
 mask_pii(ssn, 'ssn') as "ssn",
 mask_pii(cast(date_of_birth as VARCHAR), 'dob') as "date_of_birth"
from users_sensitive_data

-- Create Iceberg Sink
CREATE SINK sink_users_sensitive_data_masking FROM users_sensitive_data_masking
WITH (
    connector = 'iceberg',
    type = 'upsert',
    primary_key = 'id',
    s3.endpoint = 'http://minio:9000',
    s3.region = 'us-east-1',
    s3.access.key = 'admin',
    s3.secret.key = 'password',
    s3.path.style.access = 'true',
    catalog.type = 'rest',
    catalog.uri = 'http://amoro:1630/api/iceberg/rest',
    catalag.name = 'icelake',
    warehouse.path = 'icelake',
    database.name = 'warehouse',
    table.name = 'users_sensitive_data_masking',
    create_table_if_not_exists = TRUE
);

Now let us check our data in Apache Iceberg using Starrocks as query engine. Create external catalog in Starrocks with script :

CREATE EXTERNAL CATALOG iceberg_catalog
PROPERTIES
(
 "type"  =  "iceberg", 
    "iceberg.catalog.uri"  = "http://amoro:1630/api/iceberg/rest", 
    "iceberg.catalog.type"  =  "rest",   
    "iceberg.catalog.warehouse" = "icelake",
    "aws.s3.use_instance_profile" = "false",
    "aws.s3.access_key" = "admin",
    "aws.s3.secret_key" = "password",
    "aws.s3.region" = "us-east-1",
    "aws.s3.enable_ssl" = "false",
    "aws.s3.enable_path_style_access" = "true",
    "aws.s3.endpoint" = "http://minio:9000"
);

set catalog iceberg_catalog;
show databases;
use warehouse;
show tables;
select * from users_sensitive_data_masking;

Taraaa, you can check that your data has been masked

Perfect.

Thanks for reading my article. I hope you enjoyed it, and see you in the next one, Cheerrssss 🥂