Program: Save Athena query data to DynamoDB Table

1. Import Package

import boto3
import time
import csv
import io

2. Setup required configuration object

# Configuration
DATABASE_NAME = "your_database"
TABLE_NAME = "your_dynamodb_table"
S3_OUTPUT_BUCKET = "your-athena-query-results-bucket"
S3_OUTPUT_PREFIX = "athena_results/"
ATHENA_QUERY = "SELECT id, name, age FROM your_athena_table LIMIT 10;"

3. run the athena query

def run_athena_query():
    """Submit Athena query and return the query execution ID"""
    response = athena_client.start_query_execution(
        QueryString=ATHENA_QUERY,
        QueryExecutionContext={"Database": DATABASE_NAME},
        ResultConfiguration={"OutputLocation": f"s3://{S3_OUTPUT_BUCKET}/{S3_OUTPUT_PREFIX}"}
    )
    return response['QueryExecutionId']

4. the method help to get execution id and pass the result to next stage

def wait_for_query_completion(query_execution_id):
    """Wait for Athena query to complete"""
    while True:
        response = athena_client.get_query_execution(QueryExecutionId=query_execution_id)
        status = response["QueryExecution"]["Status"]["State"]
        if status in ["SUCCEEDED", "FAILED", "CANCELLED"]:
            return status
        time.sleep(2)  # Wait before checking again

5. Fetch results from Athena query and return as a list of dictionaries

Retrieve the query results from Amazon S3 (where Athena stores output).

def fetch_query_results(query_execution_id):
    """Fetch results from Athena query and return as a list of dictionaries"""
    query_results_path = f"{S3_OUTPUT_PREFIX}{query_execution_id}.csv"
    
    # Read CSV data from S3
    obj = s3_client.get_object(Bucket=S3_OUTPUT_BUCKET, Key=query_results_path)
    content = obj['Body'].read().decode('utf-8')
    
    # Parse CSV
    data = []
    csv_reader = csv.DictReader(io.StringIO(content))
    for row in csv_reader:
        data.append(row)
    return data

6. Write query results to DynamoDB

def write_to_dynamodb(data):
    """Write query results to DynamoDB"""
    table = dynamodb.Table(TABLE_NAME)
    for row in data:
        table.put_item(Item=row)
    print("Data successfully written to DynamoDB")

2. Main method to process and starting point of program

Parse and write the results to DynamoDB using the Boto3 DynamoDB client

def main():
    query_id = run_athena_query()
    print(f"Executing Athena Query: {query_id}")

    status = wait_for_query_completion(query_id)
    if status != "SUCCEEDED":
        print(f"Query failed with status: {status}")
        return

    data = fetch_query_results(query_id)
    if data:
        write_to_dynamodb(data)
    else:
        print("No data retrieved from Athena query.")

Full program

import boto3
import time
import csv
import io

# AWS clients
athena_client = boto3.client('athena', region_name='us-east-1')
s3_client = boto3.client('s3', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')

# Configuration
DATABASE_NAME = "your_database"
TABLE_NAME = "your_dynamodb_table"
S3_OUTPUT_BUCKET = "your-athena-query-results-bucket"
S3_OUTPUT_PREFIX = "athena_results/"
ATHENA_QUERY = "SELECT id, name, age FROM your_athena_table LIMIT 10;"


def run_athena_query():
    """Submit Athena query and return the query execution ID"""
    response = athena_client.start_query_execution(
        QueryString=ATHENA_QUERY,
        QueryExecutionContext={"Database": DATABASE_NAME},
        ResultConfiguration={"OutputLocation": f"s3://{S3_OUTPUT_BUCKET}/{S3_OUTPUT_PREFIX}"}
    )
    return response['QueryExecutionId']


def wait_for_query_completion(query_execution_id):
    """Wait for Athena query to complete"""
    while True:
        response = athena_client.get_query_execution(QueryExecutionId=query_execution_id)
        status = response["QueryExecution"]["Status"]["State"]
        if status in ["SUCCEEDED", "FAILED", "CANCELLED"]:
            return status
        time.sleep(2)  # Wait before checking again


def fetch_query_results(query_execution_id):
    """Fetch results from Athena query and return as a list of dictionaries"""
    query_results_path = f"{S3_OUTPUT_PREFIX}{query_execution_id}.csv"
    
    # Read CSV data from S3
    obj = s3_client.get_object(Bucket=S3_OUTPUT_BUCKET, Key=query_results_path)
    content = obj['Body'].read().decode('utf-8')
    
    # Parse CSV
    data = []
    csv_reader = csv.DictReader(io.StringIO(content))
    for row in csv_reader:
        data.append(row)
    return data


def write_to_dynamodb(data):
    """Write query results to DynamoDB"""
    table = dynamodb.Table(TABLE_NAME)
    for row in data:
        table.put_item(Item=row)
    print("Data successfully written to DynamoDB")


def main():
    query_id = run_athena_query()
    print(f"Executing Athena Query: {query_id}")

    status = wait_for_query_completion(query_id)
    if status != "SUCCEEDED":
        print(f"Query failed with status: {status}")
        return

    data = fetch_query_results(query_id)
    if data:
        write_to_dynamodb(data)
    else:
        print("No data retrieved from Athena query.")


if __name__ == "__main__":
    main()