Objective:

• Build a Python script to automate data ingestion:

  1. Query data from a SQLite database.

  2. Save the queried data as a CSV file locally.

  3. Upload the CSV file to an AWS S3 bucket.

Step-by-Step Instructions

1. Prepare the Environment

Install Required Libraries:
bash
Copy code
pip install pandas boto3 sqlite3

1. • Download the Sample Database:

   • Use the provided employees.db SQLite database or download a similar dataset from Kaggle (e.g., Employee Data).

   • Set Up AWS S3:

   • Create an S3 bucket if you haven’t already (e.g., my-data-bucket).

   • Configure your AWS credentials as described in Lesson 2.

2. Write the Data Ingestion Script

python

Copy code

import sqlite3

import pandas as pd

import boto3

from botocore.exceptions import NoCredentialsError

# Step 1: Query Data from SQLite Database

def fetch_data_from_db(db_path, query):

    try:

        # Connect to the database

        connection = sqlite3.connect(db_path)

        print("Connected to SQLite database.")

        # Execute query and fetch data

        data = pd.read_sql_query(query, connection)

        # Close connection

        connection.close()

        print("Data fetched successfully.")

        return data

    except Exception as e:

        print(f"Error querying the database: {e}")

        return None

# Step 2: Save Data Locally as a CSV File

def save_data_as_csv(data, file_path):

    try:

        data.to_csv(file_path, index=False)

        print(f"Data saved locally as '{file_path}'.")

    except Exception as e:

        print(f"Error saving data as CSV: {e}")

# Step 3: Upload the CSV File to AWS S3

def upload_to_s3(bucket_name, file_name, s3_key):

    s3 = boto3.client('s3')

    try:

        s3.upload_file(file_name, bucket_name, s3_key)

        print(f"File '{file_name}' uploaded to S3 bucket '{bucket_name}' as '{s3_key}'.")

    except FileNotFoundError:

        print(f"File '{file_name}' not found.")

    except NoCredentialsError:

        print("AWS credentials not found.")

    except Exception as e:

        print(f"Error uploading to S3: {e}")

# Main Function

if __name__ == "__main__":

    # Database and S3 configuration

    db_path = "employees.db"

    query = "SELECT * FROM employees"

    local_csv = "employees_data.csv"

    bucket_name = "my-data-bucket"

    s3_key = "ingested_data/employees_data.csv"

    # Fetch data from database

    data = fetch_data_from_db(db_path, query)

    # Save data locally as CSV

    if data is not None:

        save_data_as_csv(data, local_csv)

        # Upload CSV to S3

        upload_to_s3(bucket_name, local_csv, s3_key)

Explanation of the Script

1. Query Data from SQLite:

   • Connects to the SQLite database.

   • Fetches data using a SQL query.

   • Converts the result into a Pandas DataFrame for easy manipulation.

2. Save as CSV:

   • Saves the DataFrame as a CSV file locally for review or future use.

3. Upload to AWS S3:

   • Uses boto3 to upload the CSV file to the specified S3 bucket and key.

Assignment

Task:

• Modify the script to:

  • Query a different table or dataset.

  • Save the file with a different name (e.g., new_employees_data.csv).

  • Upload it to a new folder in the same S3 bucket (e.g., new_ingested_data/).

Deliverable:

• A Python script that successfully ingests data and uploads it to S3.

Resources

1. Courses:

   • Udemy: Python Programming Masterclass

   • Udemy: Data Analysis with Pandas

2. Datasets:

   • Kaggle Employee Data

   • Sample SQLite database provided in the course.

3. Documentation:

   • SQLite Official Documentation

   • Pandas Documentation

   • AWS boto3 Documentation

Expected Output

1. A local CSV file (e.g., employees_data.csv) saved with the queried data.

2. The CSV file uploaded to the S3 bucket in the specified folder.

Learning Outcomes

• Automate the process of querying data from a database.

• Save data in a structured format (CSV) for further analysis.

• Seamlessly upload data to AWS S3 for cloud-based storage and workflows.

Objective

• Modify your existing script to:

  1. Query a new dataset from a database.

  2. Save the queried data as a CSV file.

  3. Upload the CSV file to a specified folder in your S3 bucket.

Steps to Complete the Assignment

1. Select or Create a New Dataset

• Option 1: Use a Provided Dataset:

  • Download a sample dataset, such as the Kaggle Employee Data.

  • If the dataset is in CSV format, import it into a SQLite database for consistency with the activity.

Example: Importing a CSV into SQLite:
bash
Copy code
sqlite3 new_employees.db

.mode csv

.import new_employees.csv employees

.quit

• Option 2: Create Your Own Dataset:

  • Use any dataset with similar fields (e.g., name, role, department).

  • Add the data to a new table in the SQLite database.

2. Modify the Script

• Update the query to extract data from the new table.

• Change the file name for the CSV and the destination key in the S3 bucket.

Example Script

python

Copy code

import sqlite3

import pandas as pd

import boto3

from botocore.exceptions import NoCredentialsError

# Function to fetch data from the new dataset

def fetch_new_data_from_db(db_path, query):

    try:

        connection = sqlite3.connect(db_path)

        print("Connected to SQLite database.")

        # Execute the query

        data = pd.read_sql_query(query, connection)

        connection.close()

        print("New data fetched successfully.")

        return data

    except Exception as e:

        print(f"Error querying the database: {e}")

        return None

# Function to save data locally as a CSV file

def save_new_data_as_csv(data, file_path):

    try:

        data.to_csv(file_path, index=False)

        print(f"New data saved locally as '{file_path}'.")

    except Exception as e:

        print(f"Error saving data as CSV: {e}")

# Function to upload the new CSV file to S3

def upload_new_csv_to_s3(bucket_name, file_name, s3_key):

    s3 = boto3.client('s3')

    try:

        s3.upload_file(file_name, bucket_name, s3_key)

        print(f"File '{file_name}' uploaded to S3 bucket '{bucket_name}' as '{s3_key}'.")

    except FileNotFoundError:

        print(f"File '{file_name}' not found.")

    except NoCredentialsError:

        print("AWS credentials not found.")

    except Exception as e:

        print(f"Error uploading to S3: {e}")

# Main function

if __name__ == "__main__":

    # Database and S3 configuration

    db_path = "new_employees.db"  # New database file

    query = "SELECT * FROM employees"  # New query

    local_csv = "new_employees_data.csv"  # New CSV file name

    bucket_name = "my-data-bucket"

    s3_key = "new_ingested_data/new_employees_data.csv"  # New folder/key in S3

    # Fetch new data from the database

    new_data = fetch_new_data_from_db(db_path, query)

    # Save the new data locally

    if new_data is not None:

        save_new_data_as_csv(new_data, local_csv)

        # Upload the new CSV to S3

        upload_new_csv_to_s3(bucket_name, local_csv, s3_key)

Testing and Verification

1. Run the Script

Execute the script to ensure that the new dataset is successfully queried, saved, and uploaded to S3:

bash

Copy code

python upload_new_dataset.py

2. Verify the Output

• Local File:

  • Check that the new CSV file (new_employees_data.csv) is created locally with the correct data.

• S3 Bucket:

  • Log in to the AWS Management Console.

  • Navigate to your S3 bucket and locate the new_ingested_data/ folder.

  • Verify that the new file (new_employees_data.csv) is present.

Deliverables

1. Script:

   • Submit the updated Python script.

2. CSV File:

   • Provide the name of the locally saved CSV file.

3. S3 Confirmation:

   • Screenshot or log showing successful upload to S3.

Additional Challenge

• Implement logging to track each step of the script execution.

• Add error handling to manage cases like:

  • Missing data.

  • Invalid queries.

  • Connection issues with S3.

Learning Outcomes

• Modify and adapt Python scripts for different datasets.

• Save and upload data to S3 programmatically.

• Ensure automation of data ingestion workflows for scalability.