עיבוד נתונים בעננים עם DuckDB ו-AWS S3

# Install required packages for cloud integration

pip install "duckdb>=0.8.0" pyarrow pandas boto3 requests

import duckdb

import os

​

# Initialize DuckDB with cloud support

conn = duckdb.connect(':memory:')

conn.execute("INSTALL httpfs;")

conn.execute("LOAD httpfs;")

​

# Secure AWS configuration

conn.execute("""

    SET s3_region='your-region';

    SET s3_access_key_id='your-access-key';

    SET s3_secret_access_key='your-secret-key';

""")

import duckdb

import pandas as pd

​

# Create sample data to demonstrate parquet processing

sample_data = pd.DataFrame({

    'name': ['John Smith', 'Jane Doe', 'Bob Wilson', 'Alice Brown'],

    'email': ['[email protected]', '[email protected]', '[email protected]', '[email protected]'],

    'phone': ['123-456-7890', '234-567-8901', '345-678-9012', '456-789-0123'],

    'ssn': ['123-45-6789', '234-56-7890', '345-67-8901', '456-78-9012'],

    'address': ['123 Main St', '456 Oak Ave', '789 Pine Rd', '321 Elm Dr'],

    'salary': [75000, 85000, 65000, 95000]  # Non-sensitive data

})

def demonstrate_parquet_processing():

    # Create a DuckDB connection

    conn = duckdb.connect(':memory:')

    # Save sample data as parquet

    sample_data.to_parquet('sample_data.parquet')

    # Define sensitive columns to mask

    sensitive_cols = ['email', 'phone', 'ssn']

    # Process the parquet file with masking

    query = f"""

    CREATE TABLE masked_data AS

    SELECT 

        -- Mask name: keep first letter of first and last name

        regexp_replace(name, '([A-Z])[a-z]+ ([A-Z])[a-z]+', '\1*** \2***') as name,

        -- Mask email: hide everything before @

        regexp_replace(email, '([a-zA-Z0-9._%+-]+)(@.*)', '****\2') as email,

        -- Mask phone: show only last 4 digits

        regexp_replace(phone, '[0-9]{3}-[0-9]{3}-', '***-***-') as phone,

        -- Mask SSN: show only last 4 digits

        regexp_replace(ssn, '[0-9]{3}-[0-9]{2}-', '***-**-') as ssn,

        -- Mask address: show only street type

        regexp_replace(address, '[0-9]+ [A-Za-z]+ ', '*** ') as address,

        -- Keep non-sensitive data as is

        salary

    FROM read_parquet('sample_data.parquet');

    """

Original Data:

=============

         name                   email           phone          ssn        address  salary

0  John Smith    [email protected]   123-456-7890  123-45-6789   123 Main St   75000

1   Jane Doe    [email protected]   234-567-8901  234-56-7890   456 Oak Ave   85000

2  Bob Wilson         [email protected]     345-678-9012  345-67-8901   789 Pine Rd   65000

3 Alice Brown      [email protected]     456-789-0123  456-78-9012   321 Elm Dr    95000

​

Masked Data:

===========

      name                email           phone          ssn     address  salary

0  J*** S***    ****@email.com     ***-***-7890  ***-**-6789   *** St   75000

1  J*** D***  ****@company.com     ***-***-8901  ***-**-7890   *** Ave   85000

2  B*** W***      ****@email.net   ***-***-9012  ***-**-8901   *** Rd   65000

3  A*** B***      ****@org.com     ***-***-0123  ***-**-9012   *** Dr    95000

# Show first letter only

"[email protected]" → "j***@email.com"

​

# Show domain only

"[email protected]" → "****@email.com"

​

# Show first and last letter

"[email protected]" → "j*********[email protected]"

# Last 4 digits only

"123-456-7890" → "***-***-7890"

​

# First 3 digits only

"123-456-7890" → "123-***-****"

​

# Middle digits only

"123-456-7890" → "***-456-****"

# Initials only

"John Smith" → "J.S."

​

# First letter of each word

"John Smith" → "J*** S***"

​

# Fixed length masking

"John Smith" → "XXXX XXXXX"

def process_partitioned_data(base_path, partition_column, sensitive_columns):

    """

    Process partitioned data efficiently

    Parameters:

    - base_path: Base path to partitioned data

    - partition_column: Column used for partitioning (e.g., 'date')

    - sensitive_columns: List of columns to mask

    """

    conn = duckdb.connect(':memory:')

    try:

        # 1. List all partitions

        query = f"""

        WITH partitions AS (

            SELECT DISTINCT {partition_column}

            FROM read_parquet('{base_path}/*/*.parquet')

        )

        SELECT * FROM partitions;

        """

sample_data/

├── date=2024-01-01/

│   └── data.parquet

├── date=2024-01-02/

│   └── data.parquet

└── date=2024-01-03/

    └── data.parquet

# Optimize for performance

conn.execute("""

    SET partial_streaming=true;

    SET threads=4;

    SET memory_limit='4GB';

""")

def robust_s3_read(s3_path, max_retries=3):

    """

    Implement reliable S3 data reading with retries.

    Parameters:

    - s3_path: Path to S3 data

    - max_retries: Maximum retry attempts

    """

    for attempt in range(max_retries):

        try:

            return conn.execute(f"SELECT * FROM read_parquet('{s3_path}')")

        except Exception as e:

            if attempt == max_retries - 1:

                raise

            time.sleep(2 ** attempt)  # Exponential backoff

# Efficient data storage with compression

conn.execute("""

    COPY (SELECT * FROM masked_data)

    TO 's3://output-bucket/masked_data.parquet'

    (FORMAT 'parquet', COMPRESSION 'ZSTD');

""")