import duckdb
import boto3
import json
from typing import List, Dict

# Initialize S3 client
s3 = boto3.client('s3')

def lambda_handler(event, context):
    """
    DuckDB on AWS Lambda Example:
    Read Parquet files directly from S3 and analyze
    """
    
    # Read Parquet file from S3
    bucket = event['bucket']
    key = event['key']
    
    # Method 1: Read directly from S3 (requires httpfs extension)
    con = duckdb.connect()
    
    # Method 2: Download first, then process (suitable for small files)
    local_path = '/tmp/data.parquet'
    s3.download_file(bucket, key, local_path)
    
    # Analyze with DuckDB
    result = con.execute(f'''
        SELECT 
            customer_id,
            SUM(amount) as total_amount,
            AVG(amount) as avg_amount,
            COUNT(*) as order_count
        FROM read_parquet('{local_path}')
        GROUP BY customer_id
        ORDER BY total_amount DESC
        LIMIT 10
    ''').fetchdf()
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'top_customers': result.to_dict('records')
        })
    }

duckdb==1.5.5
boto3==1.35.0
pandas==2.2.0

# Create deployment package
mkdir -p deploy
cd deploy

# Install dependencies
pip install duckdb boto3 pandas -t .

# Package
zip -r ../duckdb-lambda.zip .

# Upload to Lambda
aws lambda create-function \
    --function-name duckdb-analyzer \
    --runtime python3.11 \
    --role arn:aws:iam::YOUR_ACCOUNT:role/lambda-role \
    --handler lambda_handler.lambda_handler \
    --zip-file fileb://duckdb-lambda.zip \
    --timeout 300 \
    --memory-size 1024

{
  "Events": ["s3:ObjectCreated:*"],
  "Filter": {
    "Key": {
      "S3RegExp": ".*\\.parquet$"
    }
  }
}

-- Register DuckDB engine in Athena
CREATE FUNCTION duckdb AS 'com.amazonaws.athena.connectors.duckdb.DuckDBFunctionSchema'
USING CLASSPATH '/path/to/duckdb-connector.jar';

-- Use DuckDB engine to query Parquet files on S3
SELECT 
    customer_id,
    SUM(amount) as total_amount,
    AVG(amount) as avg_amount
FROM s3_bucket.my_data.parquet
GROUP BY customer_id
ORDER BY total_amount DESC;

import duckdb
import pandas as pd
from sagemaker import get_execution_role
import boto3

# Initialize DuckDB
con = duckdb.connect()

# Read S3 data
s3_uri = 's3://your-bucket/data.parquet'
con.execute(f"CREATE TABLE data AS SELECT * FROM read_parquet('{s3_uri}')")

# Data preprocessing
preprocessed = con.execute("""
    SELECT 
        *,
        CASE 
            WHEN amount > 1000 THEN 'high'
            WHEN amount > 500 THEN 'medium'
            ELSE 'low'
        END as amount_tier
    FROM data
""").fetchdf()

import time
import pandas as pd

# Test data: 10GB Parquet file
data_path = 's3://your-bucket/large_dataset.parquet'

# Method 1: Pandas (requires download)
start = time.time()
df_pandas = pd.read_parquet(data_path)
pandas_time = time.time() - start
print(f"Pandas: {pandas_time:.2f} seconds")

# Method 2: DuckDB (direct S3 read)
start = time.time()
con = duckdb.connect()
df_duckdb = con.execute(f"SELECT * FROM read_parquet('{data_path}')").fetchdf()
duckdb_time = time.time() - start
print(f"DuckDB: {duckdb_time:.2f} seconds")

# Performance improvement
improvement = (pandas_time - duckdb_time) / pandas_time * 100
print(f"Performance improvement: {improvement:.1f}%")

import duckdb
import boto3
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText

class EcommerceAnalyzer:
    def __init__(self, bucket_name: str):
        self.bucket = bucket_name
        self.con = duckdb.connect()
        self.s3 = boto3.client('s3')
    
    def ingest_daily_data(self, date: str):
        """Daily data ingestion"""
        s3_key = f"sales/{date}/data.parquet"
        
        # Use DuckDB to read S3 directly (requires httpfs extension)
        self.con.execute(f"""
            CREATE TABLE daily_sales_{date} AS
            SELECT * 
            FROM read_parquet('s3://{self.bucket}/{s3_key}')
        """)
    
    def generate_report(self, date: str) -> dict:
        """Generate sales report"""
        table_name = f"daily_sales_{date}"
        
        report = self.con.execute(f"""
            WITH daily_stats AS (
                SELECT 
                    DATE(order_date) as order_date,
                    COUNT(*) as total_orders,
                    SUM(amount) as total_revenue,
                    AVG(amount) as avg_order_value,
                    COUNT(DISTINCT customer_id) as unique_customers,
                    SUM(CASE WHEN amount > 1000 THEN 1 ELSE 0 END) as high_value_orders,
                    SUM(CASE WHEN amount < 10 THEN 1 ELSE 0 END) as suspicious_orders
                FROM {table_name}
                GROUP BY DATE(order_date)
            ),
            category_performance AS (
                SELECT 
                    category,
                    SUM(amount) as category_revenue,
                    COUNT(*) as category_orders,
                    AVG(amount) as avg_category_value
                FROM {table_name}
                GROUP BY category
                ORDER BY category_revenue DESC
                LIMIT 10
            ),
            anomaly_detection AS (
                SELECT 
                    order_id,
                    customer_id,
                    amount,
                    'high_value' as anomaly_type
                FROM {table_name}
                WHERE amount > 10000
                UNION ALL
                SELECT 
                    order_id,
                    customer_id,
                    amount,
                    'suspicious' as anomaly_type
                FROM {table_name}
                WHERE amount < 10
            )
            SELECT 
                (SELECT * FROM daily_stats) as daily_stats,
                (SELECT * FROM category_performance) as top_categories,
                (SELECT * FROM anomaly_detection) as anomalies
        """).fetchall()
        
        return {
            'date': date,
            'stats': report[0],
            'categories': report[1],
            'anomalies': report[2]
        }
    
    def send_email_report(self, report: dict, recipients: list):
        """Send email report"""
        subject = f"Daily Sales Report - {report['date']}"
        
        body = f"""
        Daily Sales Report - {report['date']}
        
        === Key Metrics ===
        Total Orders: {report['stats']['total_orders']:,}
        Total Revenue: ${report['stats']['total_revenue']:,.2f}
        Average Order Value: ${report['stats']['avg_order_value']:.2f}
        Unique Customers: {report['stats']['unique_customers']:,}
        
        === Anomalies ===
        High-value Orders: {report['stats']['high_value_orders']}
        Suspicious Orders: {report['stats']['suspicious_orders']}
        
        === Top 10 Categories ===
        """
        
        for i, cat in enumerate(report['categories'], 1):
            body += f"{i}. {cat['category']}: ${cat['category_revenue']:,.2f}\n"
        
        # Email sending logic...
        print(body)

AWSTemplateFormatVersion: '2010-09-09'
Resources:
  DailyReportFunction:
    Type: AWS::Lambda::Function
    Properties:
      Handler: index.lambda_handler
      Runtime: python3.11
      Role: !GetAtt LambdaRole.Arn
      Code:
        ZipFile: |
          import duckdb
          import boto3
          from datetime import datetime, timedelta
          
          def lambda_handler(event, context):
              yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
              analyzer = EcommerceAnalyzer('your-bucket')
              analyzer.ingest_daily_data(yesterday)
              report = analyzer.generate_report(yesterday)
              analyzer.send_email_report(report, ['[email protected]'])
              return {'statusCode': 200}
      Timeout: 300
      MemorySize: 1024

  DailyTrigger:
    Type: AWS::Events::Rule
    Properties:
      ScheduleExpression: 'cron(0 9 * * ? *)'
      State: ENABLED
      Targets:
        - Arn: !GetAtt DailyReportFunction.Arn
          Id: DailyReport