Skyflow EMR Data Tokenization Pipeline

Complete EMR-based data tokenization solution using Skyflow's vault for secure SSN processing. Supports both Python (PySpark) and Scala (Spark) implementations on the same cluster.

Overview

This project provides a complete AWS EMR pipeline for:

Generating mock patient data with SSNs
Tokenizing SSNs using Skyflow vault
Detokenizing tokens back to original values
Performance metrics tracking for both Python and Scala implementations

Key Features:

Dual implementation (Python & Scala) on single cluster
Centralized JSON configuration
Idempotent operations (safe retries)
Comprehensive performance metrics
Language-specific S3 output paths

Prerequisites

Local Machine

AWS CLI configured with credentials
Python 3.x (for config parsing)

SBT (Scala Build Tool) - only for Scala jobs

# macOS
brew install sbt

# Linux
# See https://www.scala-sbt.org/download.html

AWS Account

EMR permissions
S3 permissions
IAM role creation permissions

Skyflow Account

Vault ID and URL
API key or service account credentials
Table with SSN column configured

Quick Start

Complete Setup from Scratch

# 1. DEPLOY INFRASTRUCTURE (~10 minutes)
cd scripts && ./deploy.sh

# 2a. RUN PYTHON PIPELINE (no build needed)
./run.sh pipeline 1000

# 2b. RUN SCALA PIPELINE (auto-builds with --build flag)
./run.sh --lang scala --build pipeline 1000

# 3. CLEANUP WHEN DONE
./cleanup.sh

Command Reference

1. Initial Deployment

cd scripts
./deploy.sh

What it does:

Creates S3 bucket: skyflow-emr-<timestamp>
Creates/verifies IAM roles for EMR
Uploads Python scripts and config to S3
Creates EMR cluster (5x m5.xlarge nodes)
Waits for cluster to be ready (~10 minutes)
Generates local config: config/skyflow_config.deployed.json

Output:

Cluster ID: j-XXXXXXXXXXXXX
S3 Bucket:  s3://skyflow-emr-1234567890
Region:     us-east-2

2. Run Jobs

All jobs are executed using the unified run.sh script:

./run.sh [OPTIONS] COMMAND [ARGS]

Commands:
  generate NUM_ROWS      Generate mock data only
  tokenize NUM_ROWS      Generate + tokenize
  detokenize             Detokenize existing tokens
  pipeline NUM_ROWS      Full pipeline: generate → tokenize → detokenize

Options:
  --lang python|scala    Language implementation (default: python)
  --build                Rebuild scripts/jars before running
  --help, -h             Show help

Python Examples (No Build Required)

# Full pipeline
./run.sh pipeline 1000                    # 1,000 records
./run.sh pipeline 50000                   # 50,000 records

# Individual steps
./run.sh generate 1000                    # Generate data only
./run.sh tokenize 1000                    # Generate + tokenize
./run.sh detokenize                       # Detokenize existing tokens

Scala Examples (Auto-build with --build)

# Full pipeline (auto-build JAR)
./run.sh --lang scala --build pipeline 1000

# Individual steps
./run.sh --lang scala --build tokenize 5000
./run.sh --lang scala detokenize

# Without rebuild (faster, use after first build)
./run.sh --lang scala pipeline 1000

3. Code Updates

Update Python Scripts

# Edit files in jobs/python/*.py.template
# Then rebuild:
cd scripts
./run.sh --build pipeline 1000

Update Scala Code

# Edit files in jobs/scala/*.scala.template
# Then rebuild:
cd scripts
./run.sh --lang scala --build pipeline 1000

4. Cleanup

./cleanup.sh

What it does:

Terminates EMR cluster
Deletes S3 bucket and all contents
Prompts for IAM role deletion (optional)

⚠️ Warning: This is destructive - all data will be deleted!

Configuration

Main Config File: `config/skyflow_config.json`

Skyflow Settings:

{
  "skyflow": {
    "vault_id": "your-vault-id",
    "vault_url": "https://your-cluster.vault.skyflowapis.com",
    "table_name": "ssn",
    "column_name": "ssn",
    "api_key_file": "/home/hadoop/skyflow_api_key.txt",
    "credentials_file": "/home/hadoop/skyflow_creds.json"
  }
}

Processing Settings:

{
  "processing": {
    "batch_size": 300,        // Records per API call
    "max_workers": 8,         // Concurrent threads per partition
    "num_partitions": 1       // Spark partitions (1 for testing)
  }
}

Deployment Settings:

{
  "deployment": {
    "region": "us-east-2",                    // AWS region (editable)
    "cluster_id": "{{CLUSTER_ID}}",           // Auto-populated by deploy.sh
    "timestamp": "{{TIMESTAMP}}"              // Auto-populated by deploy.sh
  }
}

Credential Files

API Key: credentials/credentials_api_key.txt

your-skyflow-api-key-here

Service Account (Alternative): credentials/credentials.json

{
  "clientID": "your-client-id",
  "clientName": "your-client-name",
  "keyID": "your-key-id",
  "tokenURI": "https://manage.skyflowapis.com/v1/auth/sa/oauth/token",
  "privateKey": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
}

Output Locations

Python Outputs

s3://BUCKET/
├── data/
│   ├── generated/                       # Mock patient data (shared)
│   ├── tokenized_python/                # Tokenized SSNs (Python)
│   ├── tokenization_mapping_python/     # Original → Token mapping
│   └── detokenized_python/              # Detokenized results
└── metrics/
    ├── generate_data_metrics/           # Generation metrics
    ├── tokenization_metrics_python/     # Tokenization metrics
    └── detokenization_metrics_python/   # Detokenization metrics

Scala Outputs

s3://BUCKET/
├── data/
│   ├── generated/                       # Mock patient data (shared)
│   ├── tokenized_scala/                 # Tokenized SSNs (Scala)
│   ├── tokenization_mapping_scala/      # Original → Token mapping
│   └── detokenized_scala/               # Detokenized results
└── metrics/
    ├── tokenization_metrics_scala/      # Tokenization metrics
    └── detokenization_metrics_scala/    # Detokenization metrics

Download Results

# Download Python results
aws s3 cp s3://BUCKET/data/tokenized_python/ ./output/ --recursive

# Download Scala results
aws s3 cp s3://BUCKET/data/tokenized_scala/ ./output/ --recursive

# Download metrics
aws s3 cp s3://BUCKET/metrics/ ./metrics_output/ --recursive

Troubleshooting

Check Cluster Status

source config/load_config.sh
aws emr describe-cluster --cluster-id $CLUSTER_ID --region $REGION

View Job Logs

# Get step ID from run script output, then:
aws s3 ls s3://$BUCKET_NAME/logs/$CLUSTER_ID/steps/STEP_ID/

# Download stdout
aws s3 cp s3://$BUCKET_NAME/logs/$CLUSTER_ID/steps/STEP_ID/stdout.gz - | gunzip

# Download stderr
aws s3 cp s3://$BUCKET_NAME/logs/$CLUSTER_ID/steps/STEP_ID/stderr.gz - | gunzip

Common Issues

"CLUSTER_ID not set in config"

Problem: Running build/run scripts before deployment Solution: Run ./deploy.sh first

"Scala JAR not found in S3"

Problem: Trying to run Scala jobs before building Solution: Use the --build flag:

./run.sh --lang scala --build pipeline 1000

"SBT not installed"

Problem: SBT required for Scala builds Solution:

# macOS
brew install sbt

# Linux
# See https://www.scala-sbt.org/download.html

Build Errors After Config Changes

Problem: Config changes require rebuild Solution:

# For Python
./run.sh --build pipeline 1000

# For Scala
./run.sh --lang scala --build pipeline 1000

Duplicate Records in Vault

Problem: Job failed and retried Solution: Jobs are idempotent - duplicates are handled automatically with upsert mode

Performance Comparison

Python vs Scala

Feature	Python (PySpark)	Scala Spark
Build Step	❌ None	✅ Required
Deploy Time	✅ Instant	⏱️ 2-5 min
Execution Speed	Slower (interpreted)	✅ Faster (compiled JVM)
SDK	Skyflow Python SDK 2.1.0b1	Skyflow Java SDK v2.0.0
Best For	Quick testing, development	Performance testing, production

Benchmark Results

# Test with 10,000 records
./run.sh tokenize 10000                # Python: ~45 seconds
./run.sh --lang scala tokenize 10000   # Scala: ~30 seconds

Advanced Usage

Custom Number of Records

./run.sh tokenize 50000                   # Python: 50k records
./run.sh --lang scala tokenize 50000      # Scala: 50k records

Monitor Running Jobs

# Watch cluster state
watch -n 5 'aws emr describe-cluster --cluster-id $CLUSTER_ID --query "Cluster.Status.State" --output text'

# List all steps
aws emr list-steps --cluster-id $CLUSTER_ID --region $REGION

Compare Python vs Scala Performance

# Run both with same dataset
./run.sh tokenize 10000                           # Python
./run.sh --lang scala --build tokenize 10000      # Scala

# View metrics
aws s3 cp s3://$BUCKET_NAME/metrics/tokenization_metrics_python/ - --recursive | python3 -m json.tool
aws s3 cp s3://$BUCKET_NAME/metrics/tokenization_metrics_scala/ - --recursive | python3 -m json.tool

Architecture

┌─────────────────────────────────────────────────────────┐
│                     Local Machine                        │
├─────────────────────────────────────────────────────────┤
│  deploy.sh → Creates EMR cluster + S3 bucket            │
│  run.sh → Unified script for all operations             │
│    • --build flag → Rebuilds Python/Scala artifacts     │
│    • Commands: generate, tokenize, detokenize, pipeline │
│  cleanup.sh → Destroys all resources                    │
└─────────────────────────────────────────────────────────┘
                           ↓
┌─────────────────────────────────────────────────────────┐
│                      AWS EMR Cluster                     │
├─────────────────────────────────────────────────────────┤
│  Master + 4 Core Nodes (m5.xlarge)                      │
│  Bootstrap: Downloads config + credentials from S3       │
│  Python: Runs PySpark jobs with Skyflow Python SDK      │
│  Scala: Runs Spark jobs with Skyflow Java SDK JAR       │
└─────────────────────────────────────────────────────────┘
                           ↓
┌─────────────────────────────────────────────────────────┐
│                     Skyflow Vault                        │
├─────────────────────────────────────────────────────────┤
│  Tokenize: Insert API with returnTokens=true            │
│  Detokenize: Detokenize API with plain text redaction   │
│  Idempotent: Upsert mode for safe retries               │
└─────────────────────────────────────────────────────────┘
                           ↓
┌─────────────────────────────────────────────────────────┐
│                      S3 Bucket                           │
├─────────────────────────────────────────────────────────┤
│  data/ → Generated, tokenized, detokenized datasets     │
│  metrics/ → Performance metrics (JSON)                  │
│  logs/ → EMR step logs (stdout/stderr)                  │
└─────────────────────────────────────────────────────────┘

Cost Estimation

Hourly costs (us-east-2):

EMR: 5x m5.xlarge = ~$2.50/hour
S3: Minimal (~$0.01/GB/month)
Total: ~$2.50-3.00/hour

Typical test run: 30-60 minutes = ~$2-3 total

File Structure

skyflow-emr-pipeline/
├── scripts/                             # Operational scripts
│   ├── deploy.sh                        # Deploy infrastructure
│   ├── run.sh                           # ⭐ Unified run script (use this!)
│   ├── cleanup.sh                       # Destroy resources
│   ├── install_prerequisites.sh         # One-time setup helper
│   ├── build_scala_jobs.sh              # Build Scala JARs (internal)
│   └── refresh_scripts.sh               # Update Python scripts (internal)
├── jobs/                                # Spark job source code
│   ├── python/
│   │   ├── generate_data_job.py.template
│   │   ├── skyflow_tokenize_job.py.template
│   │   └── skyflow_detokenize_job.py.template
│   └── scala/
│       ├── GenerateDataJob.scala.template
│       ├── SkyflowTokenizeJob.scala.template
│       ├── SkyflowDetokenizeJob.scala.template
│       ├── build.sbt.template
│       └── project/
│           └── assembly.sbt
├── bootstrap/                           # EMR bootstrap scripts
│   └── skyflow-bootstrap-python.sh.template
├── config/                              # Configuration files
│   ├── skyflow_config.json              # Main config (template)
│   ├── skyflow_config.deployed.json     # Deployed config (gitignored)
│   └── load_config.sh                   # Config helper script
├── credentials/                         # Skyflow credentials (gitignored)
│   ├── credentials_api_key.txt          # Your Skyflow API key
│   └── credentials.json                 # Service account (alternative)
└── README.md                            # This file

Support

For issues or questions:

Check the Troubleshooting section
Review EMR logs in S3
Check Skyflow vault logs
Verify IAM permissions

License

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

Name		Name	Last commit message	Last commit date
bootstrap		bootstrap
config		config
jobs		jobs
scripts		scripts
.gitignore		.gitignore
README.md		README.md

Uh oh!

Latest commit

History

Folders and files

Repository files navigation