Cataloging S3 Data with Glue

import boto3
import uuid
import os
from botocore.exceptions import ClientError

# Configuration
with open(".codesignal/suffix.txt") as f:
    SUFFIX = f.read().strip()
DATABASE_NAME = f'data_lake_catalog_{SUFFIX}'
CRAWLER_NAME = f'real-data-crawler-{SUFFIX}'
S3_BUCKET = f'my-data-lake-bucket-{SUFFIX}'
S3_PATH = f's3://{S3_BUCKET}/raw/user-events/'

def create_glue_crawler():
    glue = boto3.client('glue')
    sts = boto3.client('sts')
    account_id = sts.get_caller_identity()['Account']
    role_arn = f'arn:aws:iam::{account_id}:role/AWSGlueServiceRole'
    try:
        glue.create_crawler(
            Name=CRAWLER_NAME,
            Role=role_arn,
            DatabaseName=DATABASE_NAME,
            Description='Crawler for real user events data in S3',
            Targets={'S3Targets': [{'Path': S3_PATH}]},
            SchemaChangePolicy={'UpdateBehavior': 'UPDATE_IN_DATABASE', 'DeleteBehavior': 'LOG'},
            RecrawlPolicy={'RecrawlBehavior': 'CRAWL_EVERYTHING'}
        )
        print(f"Created Glue crawler: {CRAWLER_NAME}")
    except ClientError as e:
        code = e.response["Error"]["Code"]
        if code == 'AlreadyExistsException':
            print(f"Crawler {CRAWLER_NAME} already exists")
        else:
            print(f"Error creating crawler ({code}): {e.response['Error'].get('Message', str(e))}")

def run_crawler():
    """Start the crawler."""
    glue = boto3.client('glue')
    try:
        glue.start_crawler(Name=CRAWLER_NAME)
        print(f"Started crawler: {CRAWLER_NAME}")
    except ClientError as e:
        code = e.response["Error"]["Code"]
        print(f"Error running crawler ({code}): {e.response['Error'].get('Message', str(e))}")

def list_catalog_tables():
    glue = boto3.client('glue')
    try:
        response = glue.get_tables(DatabaseName=DATABASE_NAME)
        tables = response['TableList']
        print(f"\nTables discovered in {DATABASE_NAME}:")
        for table in tables:
            print(f"- {table['Name']} (Location: {table['StorageDescriptor']['Location']})")
            print("  Columns:")
            for col in table['StorageDescriptor']['Columns']:
                print(f"    - {col['Name']}: {col['Type']}")
            if table.get('PartitionKeys'):
                print("  Partition Keys:")
                for part in table['PartitionKeys']:
                    print(f"    - {part['Name']}: {part['Type']}")
            print()
    except ClientError as e:
        code = e.response["Error"]["Code"]
        print(f"Error listing tables ({code}): {e.response['Error'].get('Message', str(e))}")

if __name__ == "__main__":
    print("Connecting Glue to S3 data lake...")
    create_glue_crawler()
    run_crawler()
    # Note: The crawler runs asynchronously. It may take a few minutes to finish,
    # and tables will only appear in the catalog after the crawl completes.
    # You can re-run list_catalog_tables() later to see the discovered tables.
    list_catalog_tables()
    print("\nYour S3 data is now cataloged in Glue and ready for ETL jobs or further processing.")

Connecting Glue to S3 data lake...
Crawler real-data-crawler already exists
Started crawler: real-data-crawler

Tables discovered in data_lake_catalog:
- user_events (Location: s3://my-data-lake-bucket-<unique-suffix>/raw/user-events/year=2024/month=06/day=10/)
  Columns:
    - user_id: string
    - event_type: string
    - product_id: string
    - timestamp: string
    - session_id: string
    - revenue: double
  Partition Keys:
    - year: string
    - month: string
    - day: string