Security and Compliance for Data Engineers

Most data engineering tutorials teach you how to build pipelines that work. Almost none teach you how to build pipelines that are legal. That gap will cost you at some point — either in production when your company faces a GDPR audit, or in an interview when a hiring manager at Stripe or Venmo asks how you handle PII in your Kafka topics.

This module covers what you actually need to know as a data engineer: encryption, PII handling, access control, GDPR, and the CCPA (California Consumer Privacy Act). Not legal theory — practical decisions your pipelines must make.

1. What You Are Actually Protecting Against

Security is not abstract. As a data engineer you have three concrete threats to think about:

2. Encryption — At Rest and In Transit

Encryption is the first line of defence. There are two distinct problems: data being intercepted while it moves (in transit), and data being read from disk if storage is compromised (at rest). They require different solutions.

Encryption in transit

Any time data moves across a network — from your pipeline to a database, from Kafka producer to broker, from your API client to S3 — it must be encrypted using TLS (Transport Layer Security). Without TLS, anyone on the network path can read your data in plain text.

Encryption at rest

Encryption at rest means data stored on disk is encrypted. If someone steals the physical disk or gets unauthorized access to raw storage, they see ciphertext, not your customer records.

On all major cloud platforms, encryption at rest is enabled by default for object storage (S3, Azure Blob, GCS) and managed databases. Your job is to make sure you are using the right key type and haven't accidentally disabled it.

Column-level encryption for sensitive fields

Full-disk encryption protects you if storage is stolen. It does not protect you from a legitimate database user running SELECT email, phone FROM users. For fields like SSN numbers, phone numbers, and payment card data, you need column-level encryption — the field is stored as ciphertext in the database, and only systems with the decryption key can read the real value.

# Column-level encryption with Python (Fernet symmetric encryption)
# Use this pattern when storing sensitive fields in your data warehouse

from cryptography.fernet import Fernet
import os

# Key should come from your secrets manager (AWS Secrets Manager, Azure Key Vault)
# NEVER hardcode keys in source code
ENCRYPTION_KEY = os.environ['COLUMN_ENCRYPTION_KEY']
fernet = Fernet(ENCRYPTION_KEY.encode())

def encrypt_field(value: str) -> str:
    """Encrypt a sensitive field before writing to the database."""
    if value is None:
        return None
    return fernet.encrypt(value.encode()).decode()

def decrypt_field(encrypted_value: str) -> str:
    """Decrypt a field when it needs to be read."""
    if encrypted_value is None:
        return None
    return fernet.decrypt(encrypted_value.encode()).decode()

# In your pipeline:
row = {
    'user_id': 'U1234',
    'name': 'Jordan Lee',             # Not sensitive — store as is
    'email': encrypt_field('jordan@example.com'),   # Sensitive — encrypt
    'phone': encrypt_field('+1 512-555-0142'),       # Sensitive — encrypt
    'ssn_last4': encrypt_field('5678'),             # Sensitive — encrypt
    'city': 'Austin',             # Not sensitive — store as is
}

# Key rotation: generate new key, decrypt with old, re-encrypt with new
# This is an operational concern — document your key rotation schedule

3. PII — Identifying and Handling Personal Data

PII stands for Personally Identifiable Information — any data that can directly identify a person or, in combination with other data, identify a person. As a data engineer, your job is to know what PII your pipelines touch, where it goes, and how it is protected at every step.

What counts as PII

The four things you must do with PII in pipelines

-- Example: tagging in dbt schema.yml
models:
  - name: orders
    columns:
      - name: customer_email
        meta:
          pii: true
          pii_type: direct_identifier
          gdpr_relevant: true
          ccpa_relevant: true
      - name: customer_phone
        meta:
          pii: true
          pii_type: direct_identifier

# Data masking for dev/test environments
import hashlib
import re

def mask_email(email: str) -> str:
    """Replace real email with consistent but fake email."""
    if not email:
        return email
    hashed = hashlib.sha256(email.encode()).hexdigest()[:8]
    return f"user_{hashed}@masked.dev"

def mask_phone(phone: str) -> str:
    """Keep format, replace digits with X except last 4."""
    digits = re.sub(r'D', '', phone)
    return 'XXXXXX' + digits[-4:] if len(digits) >= 4 else 'XXXXXXXXXX'

def mask_ssn(ssn: str) -> str:
    """Standard SSN masking — show only last 4."""
    digits = re.sub(r'D', '', ssn)
    return 'XXX-XX-' + digits[-4:] if len(digits) >= 4 else 'XXX-XX-XXXX'

# Apply during the staging → dev copy process, not in production pipelines

-- Snowflake: column masking policy
CREATE OR REPLACE MASKING POLICY email_mask AS (val STRING)
RETURNS STRING ->
  CASE
    WHEN CURRENT_ROLE() IN ('DATA_ENGINEER', 'PRIVACY_ADMIN') THEN val
    ELSE CONCAT(LEFT(val, 2), '****@****.com')
  END;

-- Apply to the column
ALTER TABLE customers
  MODIFY COLUMN email
  SET MASKING POLICY email_mask;

-- Analyst sees: jo****@****.com
-- Engineer sees: jordan@freshcart.com

4. Access Control — RBAC and Least Privilege

Access control is the answer to the insider threat. The principle is simple: every user and every system gets the minimum permissions they need to do their job — nothing more. This is called least privilege.

Role-Based Access Control (RBAC)

Instead of granting permissions to individual users, you define roles (Data Engineer, Analyst, Pipeline Service Account, Admin) and assign permissions to roles. Users are assigned to roles. When someone changes jobs, you change their role — not 47 individual permissions.

Attribute-Based Access Control (ABAC)

RBAC works well when roles are stable. ABAC is more fine-grained — access is granted based on attributes of the user, the data, and the context. For example: "an analyst can read customer data only if the customer's region matches the analyst's assigned region." BigQuery, Databricks Unity Catalog, and Apache Ranger all support ABAC-style row-level and column-level security.

-- Row-level security in PostgreSQL
-- Each analyst can only see rows for their assigned region

CREATE POLICY region_isolation ON customers
  USING (region = current_setting('app.user_region'));

ALTER TABLE customers ENABLE ROW LEVEL SECURITY;

-- In your application / pipeline connection:
-- SET app.user_region = 'northeast';
-- Now queries on customers only return northeast rows

5. GDPR — What Data Engineers Need to Know

GDPR (General Data Protection Regulation) is a European Union law that came into force in 2018. It applies to any company that processes personal data of EU residents — including US companies that have EU customers. Fines go up to 4% of global annual revenue. Meta was fined €1.2 billion in 2023.

You don't need to read all 99 articles. As a data engineer, these are the 5 GDPR requirements that directly affect how you build pipelines.

Crypto-erasure — the practical way to handle deletion in data lakes

Deleting a record from a data warehouse is easy. Deleting it from an immutable data lake (S3/ADLS with versioning) is hard. The practical solution is crypto-erasure: encrypt the user's PII with a user-specific key stored in a key management service. To "delete" the user, delete their encryption key. All their encrypted data becomes permanently unreadable without modifying any files.

# Crypto-erasure pattern
# Each user's PII is encrypted with a unique per-user key
# Deletion = deleting the key from KMS

import boto3
kms = boto3.client('kms', region_name='us-east-1')

def get_or_create_user_key(user_id: str) -> str:
    """Return KMS key ARN for this user, creating if needed."""
    # In practice, store key ARN in a mapping table
    response = kms.create_key(
        Description=f'PII encryption key for user {user_id}',
        Tags=[{'TagKey': 'user_id', 'TagValue': user_id}]
    )
    return response['KeyMetadata']['KeyId']

def erase_user(user_id: str, key_id: str):
    """
    GDPR right to erasure via crypto-erasure.
    Schedules key deletion — AWS KMS minimum waiting period is 7 days.
    After deletion, all PII encrypted with this key is permanently unreadable.
    """
    kms.schedule_key_deletion(
        KeyId=key_id,
        PendingWindowInDays=7  # Minimum allowed by AWS KMS
    )
    print(f"Key for user {user_id} scheduled for deletion. PII will be unreadable in 7 days.")
    # Log this action to your audit trail
    log_audit_event('ERASURE_REQUESTED', user_id=user_id, key_id=key_id)

6. California Consumer Privacy Act (CCPA/CPRA)

The California Consumer Privacy Act took effect January 1, 2020, and was substantially expanded by the California Privacy Rights Act (CPRA), effective January 1, 2023, with enforcement beginning July 1, 2023. It is the most comprehensive US state privacy law and the de facto standard most companies build to, since a growing list of other states — Virginia, Colorado, Connecticut, Utah, and more — have passed similar laws largely modeled on it.

Key concepts in CCPA for data engineers

GDPR vs CCPA — similarities and differences

7. Compliance by Design — A Practical Checklist

Compliance bolted on after a pipeline is live is expensive and incomplete. Compliance built into the pipeline from the start is cheap and reliable. Here is the checklist you run when designing any pipeline that touches personal data.

8. Audit Logging

Audit logs answer "who did what, to which data, and when." They are your proof of compliance, your first tool in a breach investigation, and your defence in a regulatory audit. They are also one of the most commonly skipped pieces of data infrastructure.

# Minimal audit log event — write this to an immutable audit log table
# or a WORM (Write Once Read Many) log bucket

from datetime import datetime, timezone
import json

def log_audit_event(
    action: str,            # READ_PII, DELETE_RECORD, EXPORT_DATA, SCHEMA_CHANGE
    actor: str,             # user_id or service_account_name of who did it
    resource: str,          # table name, pipeline name, file path
    record_id: str = None,  # user_id or record_id affected (if applicable)
    metadata: dict = None,  # any additional context
):
    event = {
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'action': action,
        'actor': actor,
        'resource': resource,
        'record_id': record_id,
        'metadata': metadata or {},
    }
    # Write to immutable audit log — append only, no updates, no deletes
    # Options: Cloud Storage with object lock, dedicated audit table, CloudTrail, Azure Monitor
    print(json.dumps(event))  # Replace with your log sink

# Examples
log_audit_event('READ_PII', 'analyst_jordan', 'customers', metadata={'purpose': 'support_ticket_123'})
log_audit_event('DELETE_RECORD', 'privacy_admin', 'customers', record_id='U98765', metadata={'reason': 'GDPR_erasure_request'})
log_audit_event('EXPORT_DATA', 'data_engineer', 'orders', record_id='U12345', metadata={'reason': 'CCPA_access_request'})

9. Secrets Management — Never Hardcode Credentials

This is one of the most common mistakes made by junior data engineers: database passwords, API keys, and storage account keys hardcoded in Python scripts or committed to Git. A secret in your Git history is a secret that was leaked — even if you delete the commit later, it may already be in a fork or a scan.

# Fetching secrets from AWS Secrets Manager at runtime
import boto3
import json

def get_secret(secret_name: str, region: str = 'us-east-1') -> dict:
    """Fetch a secret by name. Returns dict of key-value pairs."""
    client = boto3.client('secretsmanager', region_name=region)
    response = client.get_secret_value(SecretId=secret_name)
    return json.loads(response['SecretString'])

# In your pipeline
db_secret = get_secret('freshcart/prod/postgres')
connection_string = (
    f"postgresql://{db_secret['username']}:{db_secret['password']}"
    f"@{db_secret['host']}:{db_secret['port']}/{db_secret['dbname']}"
)

# Azure equivalent: use DefaultAzureCredential + Key Vault
# from azure.keyvault.secrets import SecretClient
# from azure.identity import DefaultAzureCredential
# client = SecretClient(vault_url="https://kv-freshcart.vault.azure.net/", credential=DefaultAzureCredential())
# secret = client.get_secret("postgres-password").value