DEV Community: Felipe de Godoy

FastAPI for Data Applications: Deploying with AWS Lambda. Part III

Felipe de Godoy — Sun, 27 Apr 2025 15:26:21 +0000

In Part I and II of this blog series, we demonstrated how to create and containerize a FastAPI application to further deploy it on a Kubernetes cluster. In this follow-up post, we'll explore another approach to exposing data through an API by deploying the FastAPI application using AWS Lambda. This serverless architecture can be particularly advantageous for data engineers and data scientists seeking cost-effective, scalable solutions for their data-centric APIs.

Prerequisites

Basic knowledge of FastAPI.
An AWS account.
AWS CLI installed and configured.
serverless framework installed globally via npm:

   npm install -g serverless

Step-by-Step Guide to Exposing Data with FastAPI through AWS Lambda

1. Set Up Your FastAPI Application

Let's assume you have a FastAPI application file named main.py.

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/data")
def read_data():
    return {"message": "Welcome to FastAPI on AWS Lambda - Data Engineers/Scientists!"}

2. Create a `requirements.txt` File

Include the dependencies for your FastAPI application:

fastapi
uvicorn

3. Create a `Dockerfile` for the Serverless Framework

We'll be using Docker to package the application for AWS Lambda.

# Dockerfile
FROM amazonlinux:2

# Install Python and other requirements
RUN yum install -y python3
RUN pip3 install fastapi uvicorn

# Set the working directory
WORKDIR /var/task

# Copy application files
COPY main.py .

# FastAPI will be executed using the Uvicorn server
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

4. Serverless Framework Configuration

Initialize a new Serverless service in your project directory:

serverless create --template aws-python3 --path my-fastapi-service
cd my-fastapi-service

Replace the created handler file (handler.py) with the following content to adapt it for FastAPI:

# handler.py
import os
from mangum import Mangum
from fastapi import FastAPI

app = FastAPI()

@app.get("/data")
def read_data():
    return {"message": "Welcome to FastAPI on AWS Lambda - Data Engineers/Scientists!"}

handler = Mangum(app)

5. Update `serverless.yml` Configuration

Open serverless.yml and configure it for FastAPI:

service: my-fastapi-service

provider:
  name: aws
  runtime: python3.8
  region: us-east-1

functions:
  app:
    handler: handler.handler
    events:
      - http:
          path: /
          method: get
      - http:
          path: /data
          method: get

package:
  patterns:
    - '!**'
    - 'handler.py'

custom:
  pythonRequirements:
    dockerizePip: true

plugins:
  - serverless-python-requirements
  - serverless-wsgi

# Add Docker for packaging
custom:
  wsgi:
    app: handler.app
    packRequirements: false
  pythonRequirements:
    dockerizePip: non-linux

6. Deploy the Application

Deploy the FastAPI application to AWS Lambda using the Serverless Framework:

serverless deploy

7. Verify the Deployment

After deployment, you'll see an output with the API endpoint. You can test it via a browser or tools like curl or Postman:

curl https://<api-id>.execute-api.<region>.amazonaws.com/dev/data

This should return:

{
  "message": "Welcome to FastAPI on AWS Lambda - Data Engineers/Scientists!"
}

Conclusion

In Part I of our series, we've shown how to containerize a FastAPI application for Kubernetes. In this post, we've explored deploying a FastAPI application using AWS Lambda, providing a serverless and scalable solution. This architecture can significantly benefit data engineers and data scientists by offering a cost-effective, scalable way to expose data via APIs.

In Part II, we did dive into integrating these deployed APIs with various data sources, further leveraging the power of FastAPI in a data engineering and data science context. And now we did it in a AWS lambda

Aggregation in GROUP BY vs. Window Functions Using OVER()

Felipe de Godoy — Sun, 14 Jul 2024 23:41:24 +0000

SQL is an incredibly versatile language, indispensable for data manipulation and analysis. One of the most powerful features of SQL is its ability to perform aggregations. However, as data engineers or enthusiasts, we often need to discern between two key methods for aggregation: the GROUP BY clause and window functions using the OVER() clause. In this post series, we will dive deep into understanding the differences between these two techniques, their structures, and how they can be used to create valuable business insights.

Understanding Aggregation in GROUP BY

The GROUP BY clause in SQL is used to group rows that have the same values in specified columns into summary rows. It is a classic method for data aggregation.

Syntax:

SELECT 
    column1,
    aggregate_function(column2)
FROM table
WHERE conditions
GROUP BY column1;

Example:

SELECT 
    SalesPersonID,
    SUM(SalesAmount) AS TotalSales
FROM Sales
GROUP BY SalesPersonID;

In this example, each salesperson’s total sales are calculated by grouping rows based on SalesPersonID and summing up SalesAmount for each group.

Introducing Window Functions Using OVER()

Window functions extend SQL’s analytical capabilities by allowing calculations across sets of table rows that are related to the current row without reducing the number of rows. This means that each row retains its identity but is enriched with aggregated values based on specified criteria.

Syntax:

SELECT 
    column1,
    aggregate_function(column2) OVER (PARTITION BY column1 ORDER BY column2) AS new_column
FROM table;

Example:

SELECT 
    SalesPersonID, 
    SalesAmount,
    SUM(SalesAmount) OVER (PARTITION BY SalesPersonID) AS TotalSales
FROM Sales;

Here, SUM(SalesAmount) OVER (PARTITION BY SalesPersonID) calculates the total sales for each salesperson, adding a new column TotalSales to every row related to that salesperson.

Structure of Window Functions

Let’s break down the structure of the window function syntax: FUNCTION() OVER (PARTITION BY column1 ORDER BY column2).

1. FUNCTION()

Any standard SQL aggregate function (SUM, AVG, COUNT, MIN, MAX) or specialized window functions (ROW_NUMBER, RANK, DENSE_RANK, LAG, LEAD, FIRST_VALUE, LAST_VALUE).

2. OVER()

Defines the window over which the function operates.

3. PARTITION BY column1

Divides data into partitions. Functions operate independently within each partition.

Example:

SELECT
    SalesPersonID,
    SalesAmount,
    SUM(SalesAmount) OVER (PARTITION BY SalesPersonID) AS TotalSales
FROM Sales;

Here, PARTITION BY SalesPersonID means that SUM(SalesAmount) will be calculated separately for each salesperson, adding a total sales column without reducing rows.

4. ORDER BY column2

Orders rows within each partition. This is particularly useful for ranking, running totals, and time-series analyses.

Example:

with table_sum as 
(select 
SalesPersonID, 
sum(SalesAmount) as total_sales 
from Sales group by 1) 

select 
SalesPersonId
, Rank() over( order by total_sales desc) as ranking 
from table_sum;

In this example, ORDER BY total_sales ensures that sales are sorted by sum, in order to rank each salesperson's sell.

Comparing GROUP BY and Window Functions

Data Reduction vs. Data Preservation

GROUP BY: Aggregates data by reducing the number of rows, ideal for summarizing datasets.
Window Functions: Retains all rows, making it suitable for detailed row-level analyses.

Context and Insight

GROUP BY: Excellent for generating high-level summaries and reports.
Window Functions: Enables detailed trend analyses, running totals, and more sophisticated insights.

Business Applications

GROUP BY: Useful for dashboard summaries, quarterly reports, and grouped financial metrics.
Window Functions: Valuable for customer lifetime value analysis, stock price trends, and performance tracking over time.

Practical Performance Tips

Analyze Query Plans: Regularly use EXPLAIN to check how the database engine executes your queries.
Efficient Indexing: Always index the columns used in grouping and ordering.
Partition Wisely: Consider table partitioning for very large datasets to improve performance.

Conclusion

In summary, both GROUP BY and window functions serve unique purposes in SQL data analysis. While GROUP BY is excellent for summarizing data and producing high-level reports, window functions provide deeper insights by maintaining row-level details and allowing complex calculations over sets of related rows.

By mastering both techniques, data engineers and analysts can unlock the full potential of their data, generating insights that drive business value and optimize decision-making processes. In the next post, we’ll explore simple uses of window functions for basic aggregations like AVG, MAX, MIN, and COUNT.

Stay tuned for more deep dives into SQL capabilities! If you found this post helpful, don’t forget to share and comment below.

SQL #DataEngineering #WindowFunctions #BigData #Programming #Database

Images from https://www.boardinfinity.com/blog/window-function-in-sql/

Leveraging PySpark.Pandas for Efficient Data Pipelines

Felipe de Godoy — Thu, 04 Jul 2024 02:24:44 +0000

In the world of big data, Spark has become a pivotal tool for handling and processing large datasets efficiently. However, if you're a data scientist or a data analyst accustomed to the simplicity and power of Pandas, you might find transitioning to Spark a bit daunting. That's where the Pandas API on Spark comes in! It brings the familiar Pandas syntax to the Spark ecosystem, allowing you to leverage the distributed computing power of Spark while working with a Pandas-like interface.

Why Use Pandas API on Spark?

The Pandas API on Spark allows you to:

Handle Larger-Than-Memory Data: Work with datasets that exceed the memory capacity of a single machine.
Leverage Distributed Computing: Benefit from the parallel processing power of a Spark cluster.
Use Familiar Syntax: Transition smoothly from Pandas to Spark without having to learn a completely new API.

Setting Up Your Environment

To get started, we'll use Docker to set up a local PySpark environment. Open your terminal and run the following command:


docker run -it -p 8888:8888 jupyter/pyspark-notebook

Once the container is running, open your browser and navigate to the second link to access your PySpark environment.

Getting the Data

We'll use a dataset from Kaggle for this example. You can find the dataset here: Students Performance Dataset. Download the CSV file and place it in the appropriate location within your Docker container (you can drag it to jupyter tab in your browser).

Processing Data with Pandas API on Spark

With the environment set up and the file in the correct place, you can run the following code to read, treat, visualize, and save the data to S3.

Step 1: Import Libraries and Initialize Spark Session

!pip install boto3 plotly
import pandas as pd
import numpy as np
import pyspark.pandas as ps
from pyspark.sql import SparkSession
import boto3
spark = SparkSession.builder.appName("PandasOnSparkExample").getOrCreate()

Step 2: Read Data from CSV

columns = ['StudentID', 'Age', 'Gender', 'Ethnicity', 'ParentalEducation',
'StudyTimeWeekly', 'Absences', 'Tutoring', 'ParentalSupport',
'Extracurricular', 'Sports', 'Music', 'Volunteering', 'GPA', 'GradeClass']
psdf = ps.read_csv('Student_performance_data _.csv', names=columns, header=0)

Step 3: Exploring the Data

Check the first few rows of the dataset to ensure it's loaded correctly:

print(psdf.head())

Print column names and data types:

print(psdf.columns)
print(psdf.dtypes)

Step 4: Handling Missing Data

Handle missing data by either dropping rows with missing values:

psdf_cleaned = psdf.dropna()
print(psdf_cleaned.head())

Or filling them with a specific value:

psdf_filled = psdf.fillna(value=0)
print(psdf_filled.head())

Step 5: Data Manipulations and Insights

Group your data and apply aggregate functions:

grouped_psdf = psdf.groupby('Gender').mean()
print(grouped_psdf)

Sort your DataFrame by values:

sorted_psdf = psdf.sort_values(by='GPA', ascending=False)
print(sorted_psdf.head())

Step 6: Visualization

Plot the GPA distribution using plotly (it must be installed):

psdf['StudyTimeWeekly'].to_pandas().plot(kind='hist')

Step 7: Save as Compressed Parquet and Upload to S3

Save the DataFrame as a compressed Parquet file:

parquet_file = 'student_data.parquet.gzip'
psdf.to_parquet(parquet_file, compression='gzip')

Upload the Parquet file to S3 using boto3:

s3_bucket = 'your-s3-bucket-name'
s3_key = 'path/to/save/student_data.parquet.gzip'

# Initialize a session using Amazon S3
s3 = boto3.client('s3')

# Upload the file to S3
s3.upload_file(parquet_file, s3_bucket, s3_key)
print(f"File uploaded to s3://{s3_bucket}/{s3_key}")

Conclusion

The Pandas API on Spark bridges the gap between Pandas and Spark, offering you the best of both worlds. Whether you're handling massive datasets or looking to scale your data processing pipelines effortlessly, this API empowers you to harness the full power of Spark with the simplicity of Pandas.

Try it out and supercharge your data analytics workflow today!

For more details, you can refer to Spark's official documentation.

Happy data wrangling!

Repo: https://github.com/felipe-de-godoy/spark_with_pandas

Automating AWS Cost and Usage Report with CloudFormation

Felipe de Godoy — Thu, 27 Jun 2024 10:00:00 +0000

Automating AWS Cost and Usage Report with CloudFormation

In this blog post, we'll explore how to set up AWS Cost and Usage Report (CUR) automatically using AWS CloudFormation. This includes creating an S3 bucket for storing your reports, configuring the CUR to export data in Parquet format, and setting up Athena and Glue for querying the data. By the end of this post, you'll have a comprehensive, automated solution for managing and analyzing your AWS cost and usage data.

Prerequisites

Before diving into the CloudFormation template, you'll need:

An AWS account
AWS CLI installed and configured
Appropriate permissions to create AWS resources (S3, IAM, Athena, Glue, CUR)

Project Structure

To achieve our goal, we'll run a CloudFormation stack locally using a single YAML file. This stack will perform the following tasks:

Configure the Cost and Usage Report export
Create an Athena/Glue database

Once the table is created, we'll see an example of an Athena query for cost analysis.

By the end of this process, we will have a simplified folder structure:

cost_and_usage_report_stack
│
├── cloudformation
│ ├── MasterTemplate.yaml # Combined CloudFormation template
│
├── athena
│ ├── YourCURReport-create-table.sql # The query created by AWS
│ ├── example_query.sql # Example SQL query for Athena
│
├── README.md

Detailed Steps

1. Create the S3 Bucket

Given that the bucket name must be unique, I added my initials at the end. You will need to change this. I usually work in us-east-1 so I use this region (if you already have a bucket, skip this and update the rest) :

aws s3 mb s3://your-cur-reports-bucket-fg --region us-east-1

2. Write your Cloudformation Template

In this session, we will create with cloudformation the Cost and Usage CUR report and the Athena database to integrate it into a SQL queryable env. If you want to customize or integrate with other services like Redshift, you can adapt this YAML file.

If you have an existing database, you can skip the second part of the yaml related to Glue/Athena but remember to update any references correctly.

`cloudformation/MasterTemplate.yaml`


AWSTemplateFormatVersion: '2010-09-09'
Resources:
  CostAndUsageReportDefinition:
    Type: 'AWS::CUR::ReportDefinition'
    Properties:
      ReportName: 'YourCURReport'
      TimeUnit: 'HOURLY'
      Format: 'Parquet'
      Compression: 'Parquet'
      S3Bucket: 'your-cur-reports-bucket-fg'
      S3Prefix: 'cur-reports/'
      S3Region: 'us-east-1'
      AdditionalSchemaElements:
        - RESOURCES
      AdditionalArtifacts:
        - ATHENA
      RefreshClosedReports: true
      ReportVersioning: 'OVERWRITE_REPORT'

  GlueDatabase:
    Type: 'AWS::Glue::Database'
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseInput:
        Name: 'yourcurreport'

3. Running the CloudFormation Stack

Use AWS CLI to create the stack using MasterTemplate.yaml:

aws cloudformation create-stack --stack-name your-cur-stack --template-body file://cloudformation/MasterTemplate.yaml --capabilities CAPABILITY_NAMED_IAM

AWS processes cost data daily. In this step, you will need to wait up to 24 hours for your data to be available. If you want to create the table with my query, you can, but it'll be empty.

4. Create an Athena Table

The next day, log into your account and open your S3 bucket. The files from yesterday will be there! (So fun Huh?!)

For instance, you might find it in s3://your-cur-reports-bucket-fg/cur-reports/YourCURReport
Inside this folder, you will also find a SQL file to create the table in Athena: /YYYYMMDD-YYYYMMDD/YourCURReport-create-table.sql

In my case, that's what I found (It has many columns!):

CREATE EXTERNAL TABLE YourCURReport.your_c_u_r_report(
    identity_line_item_id STRING,
    identity_time_interval STRING,
    bill_invoice_id STRING,
    bill_invoicing_entity STRING,
    bill_billing_entity STRING,
    bill_bill_type STRING,
    bill_payer_account_id STRING,
    bill_billing_period_start_date TIMESTAMP,
    bill_billing_period_end_date TIMESTAMP,
    line_item_usage_account_id STRING,
    line_item_line_item_type STRING,
    line_item_usage_start_date TIMESTAMP,
    line_item_usage_end_date TIMESTAMP,
    line_item_product_code STRING,
    line_item_usage_type STRING,
    line_item_operation STRING,
    line_item_availability_zone STRING,
    line_item_resource_id STRING,
    line_item_usage_amount DOUBLE,
    line_item_normalization_factor DOUBLE,
    line_item_normalized_usage_amount DOUBLE,
    line_item_currency_code STRING,
    line_item_unblended_rate STRING,
    line_item_unblended_cost DOUBLE,
    line_item_blended_rate STRING,
    line_item_blended_cost DOUBLE,
    line_item_line_item_description STRING,
    line_item_tax_type STRING,
    line_item_legal_entity STRING,
    product_product_name STRING,
    product_availability_zone STRING,
    product_capacitystatus STRING,
    product_classicnetworkingsupport STRING,
    product_clock_speed STRING,
    product_current_generation STRING,
    product_dedicated_ebs_throughput STRING,
    product_ecu STRING,
    product_enhanced_networking_supported STRING,
    product_from_location STRING,
    product_from_location_type STRING,
    product_from_region_code STRING,
    product_gpu_memory STRING,
    product_group STRING,
    product_group_description STRING,
    product_instance_family STRING,
    product_instance_type STRING,
    product_instance_type_family STRING,
    product_intel_avx2_available STRING,
    product_intel_avx_available STRING,
    product_intel_turbo_available STRING,
    product_license_model STRING,
    product_location STRING,
    product_location_type STRING,
    product_marketoption STRING,
    product_max_iopsvolume STRING,
    product_max_throughputvolume STRING,
    product_max_volume_size STRING,
    product_memory STRING,
    product_network_performance STRING,
    product_normalization_size_factor STRING,
    product_operating_system STRING,
    product_operation STRING,
    product_physical_processor STRING,
    product_pre_installed_sw STRING,
    product_processor_architecture STRING,
    product_processor_features STRING,
    product_product_family STRING,
    product_region STRING,
    product_region_code STRING,
    product_servicecode STRING,
    product_servicename STRING,
    product_sku STRING,
    product_storage STRING,
    product_storage_media STRING,
    product_storage_type STRING,
    product_tenancy STRING,
    product_tiertype STRING,
    product_to_location STRING,
    product_to_location_type STRING,
    product_to_region_code STRING,
    product_transfer_type STRING,
    product_usagetype STRING,
    product_vcpu STRING,
    product_volume_api_name STRING,
    product_volume_type STRING,
    product_vpcnetworkingsupport STRING,
    pricing_rate_code STRING,
    pricing_rate_id STRING,
    pricing_currency STRING,
    pricing_public_on_demand_cost DOUBLE,
    pricing_public_on_demand_rate STRING,
    pricing_term STRING,
    pricing_unit STRING,
    reservation_amortized_upfront_cost_for_usage DOUBLE,
    reservation_amortized_upfront_fee_for_billing_period DOUBLE,
    reservation_effective_cost DOUBLE,
    reservation_end_time STRING,
    reservation_modification_status STRING,
    reservation_normalized_units_per_reservation STRING,
    reservation_number_of_reservations STRING,
    reservation_recurring_fee_for_usage DOUBLE,
    reservation_start_time STRING,
    reservation_subscription_id STRING,
    reservation_total_reserved_normalized_units STRING,
    reservation_total_reserved_units STRING,
    reservation_units_per_reservation STRING,
    reservation_unused_amortized_upfront_fee_for_billing_period DOUBLE,
    reservation_unused_normalized_unit_quantity DOUBLE,
    reservation_unused_quantity DOUBLE,
    reservation_unused_recurring_fee DOUBLE,
    reservation_upfront_value DOUBLE,
    savings_plan_total_commitment_to_date DOUBLE,
    savings_plan_savings_plan_a_r_n STRING,
    savings_plan_savings_plan_rate DOUBLE,
    savings_plan_used_commitment DOUBLE,
    savings_plan_savings_plan_effective_cost DOUBLE,
    savings_plan_amortized_upfront_commitment_for_billing_period DOUBLE,
    savings_plan_recurring_commitment_for_billing_period DOUBLE
)

PARTITIONED BY (
    year STRING,
    month STRING
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
WITH  SERDEPROPERTIES (
 'serialization.format' = '1'
) LOCATION 's3://your-cur-reports-bucket-fg/cur-reports/YourCURReport/YourCURReport/'

This file contains all the columns of the report you generated (so if you customize before, here your SQL file will be different). Open your Athena console, select the database, and run the query from the file. (You can also look more into this folder; it contains more JSONs and metadata of the process.)

Initially, if you try to directly select the table, it might return as missing. (Relax, This is expected behavior.) You need to run a command to ask Glue to search the newer partitions¹:

msck repair table yourcurreport.YourCURReport

¹ You can configure a crawler in Glue to keep updating its partitions, but it will generate costs

5. Retrieve your Data

Now you can run your query and retrieve real data about your usage!

`athena/example_query.sql`

SELECT
  line_item_usage_account_id,
  product_servicecode,
  product_product_name,
  SUM(line_item_blended_cost) as cost
FROM yourcurreport.your_cur_table
GROUP BY line_item_usage_account_id, product_servicecode, product_product_name
ORDER BY cost DESC;

This query will group costs by account and service. You can't do this in the console because it allows grouping by only one variable (and filtering by the others). In a large environment, this grouping capability can clearly show where your biggest sources of cost are.

I suggest looking into a few variables: service, usage type, region, and unblended cost. There are some query examples from AWS Well-Architected Cost Optimization.

Conclusion

In this blog post, we've automated the setup of the AWS Cost and Usage Report using a single CloudFormation template executed locally. By following these steps, you can efficiently track and manage your AWS costs. Feel free to expand this template to suit your specific requirements. If you encounter any issues with the tutorial, consider using the repo directly.

This case is interesting because, in most situations, business data is the primary focus. However, without control over cloud costs, your IT department may face significant challenges. Applying a data-driven approach to your IT challenges can be a powerful tool.

Be mindful of the costs associated with this process (storage and requests in S3, bytes read in Athena), so it's a good idea to create transformation processes that provide views aligned with your business needs, rather than looking directly into raw data.

In future posts, I will discuss good practices and frameworks for cost reduction in your environment!

GitHub Repo: https://github.com/felipe-de-godoy/cost_and_usage_report_stack

FastAPI for Data Applications: Dockerizing and Scaling Your API on Kubernetes. Part II

Felipe de Godoy — Sun, 23 Jun 2024 23:05:36 +0000

Hey there, data enthusiasts and API aficionados! 🎉 If you joined us for the first part of this series, you already have your FastAPI application running smoothly on your local machine. Today, we're going to elevate that setup by deploying it to Amazon's Elastic Kubernetes Service (EKS). Get ready to scale those RESTful routes to the cloud! 🌩️

https://media.dev.to/cdn-cgi/image/width=1000,height=420,fit=cover,gravity=auto,format=auto/https%3A%2F%2Fdev-to-uploads.s3.amazonaws.com%2Fuploads%2Farticles%2Fohr4jjoaytjmulsyxk72.png

Prerequisites

Basic knowledge of FastAPI (check out our previous post if you need a refresher).
Docker installed.
An AWS account.
kubectl and eksctl configured.

Step 1: Dockerize Your FastAPI Application

The FastAPI Application

Let's assume you have a simple FastAPI application named main.py (as shown in the initial post):

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

import uvicorn

app = FastAPI()

# Estrutura dos dados
class Item(BaseModel):
    id: int
    name: str
    description: str = None

# Banco de dados fictício (em memória)
fake_db: List[Item] = []

@app.get("/")
async def root():
    return {"message": "Hello World"}   

@app.get("/items/", response_model=List[Item])
async def get_items():
    return fake_db

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    for item in fake_db:
        if item.id == item_id:
            return item
    raise HTTPException(status_code=404, detail="Item not found")

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    # Verifica se o item já existe
    for existing_item in fake_db:
        if existing_item.id == item.id:
            raise HTTPException(status_code=400, detail="Item already exists")
    fake_db.append(item)
    return item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, updated_item: Item):
    for idx, existing_item in enumerate(fake_db):
        if existing_item.id == item_id:
            fake_db[idx] = updated_item
            return updated_item
    raise HTTPException(status_code=404, detail="Item not found")


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

In the end, we'll have a FastAPI application with a Dockerfile and Kubernetes deployment/service YAML files. Here's the file structure:

project_folder
    ├── Dockerfile
    ├── README.md
    ├── deployment.yaml
    ├── run_local.sh
    ├── main.py
    ├── requirements.txt
    ├── service.yaml

Creating the Dockerfile

Now let's build a Dockerfile to containerize our application:

# Dockerfile
# Use the official Python image from the Docker Hub

FROM tiangolo/uvicorn-gunicorn-fastapi:python3.8
COPY ./main.py /app/main.py

# Set the working directory inside the Docker container
WORKDIR /app

# Copy the requirements.txt file into the Docker image
COPY requirements.txt .

# Install the dependencies specified in requirements.txt
RUN pip install --no-cache-dir -r requirements.txt

# Copy the rest of your FastAPI code into the Docker image
# COPY . .

# Expose the port your FastAPI app runs on (default is 8000)
# EXPOSE 8000

# Command to run the FastAPI application using Uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Create the `requirements.txt` file

Feel free to include any package you could need in your process, in our case we'll need:

fastapi
uvicorn
pydantic

Building and Running the Docker Image Locally

Use the following script to build and run your Docker container locally (run_local.sh):

#!/bin/bash 
set -e 

DOCKER_IMAGE_NAME="my-fastapi-app" 

echo "Building Docker image..." 
docker build -t $DOCKER_IMAGE_NAME .

echo "Running Docker container locally..." 
docker run -d -p 8000:8000 --name my-fastapi-container $DOCKER_IMAGE_NAME 

echo "FastAPI application is now running at http://localhost:8000" 

echo "To stop the container, run: docker stop my-fastapi-container" 
echo "To remove the container, run: docker rm my-fastapi-container"

With the relevant commands in the file, you can run the script:

chmod +x run_local.sh
./run_local.sh

Step 2: Set Up an EKS Cluster

aws sts get-caller-identity
eksctl create cluster --name=minimal-cluster --region=us-east-1 --nodegroup-name=minimal-nodes --node-type=t3.micro --nodes=1 --nodes-min=1 --nodes-max=2 --node-volume-size=10 --managed

This command sets up a basic EKS cluster with minimal nodes. Super neat, right?

You can follow the progress in the console of cloudformation: eksctl creates 2 stacks, which have many assets to be built. This may take 10 minutes or longer. It's time to take a breath :)

Step 3: Push Docker Image to AWS ECR

First, create an ECR repository:

aws ecr create-repository --repository-name my-fastapi-app --region us-east-1

Next, build, tag, and push your image to ECR:

docker build -t my-fastapi-app .
docker tag my-fastapi-app:latest 836090608262.dkr.ecr.us-east-1.amazonaws.com/my-fastapi-app:latest

aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 836090608262.dkr.ecr.us-east-1.amazonaws.com
docker push 836090608262.dkr.ecr.us-east-1.amazonaws.com/my-fastapi-app:latest

Step 4: Deploy to EKS

Create a deployment.yaml:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastapi-app-deployment
spec:
  replicas: 2
  selector:
    matchLabels:
      app: fastapi-app
  template:
    metadata:
      labels:
        app: fastapi-app
    spec:
      containers:
      - name: fastapi-app
        image: 836090608262.dkr.ecr.us-east-1.amazonaws.com/my-fastapi-app:latest
        ports:
        - containerPort: 8000

And a service.yaml:

apiVersion: v1 
kind: Service 
metadata: 
  name: fastapi-app-service 
  namespace: default 
spec: 
  type: LoadBalancer 
  selector: 
    app: fastapi-app 
  ports: 
    - protocol: TCP 
      port: 80 
      targetPort: 8000

Apply them to your cluster:

kubectl apply -f deployment.yaml
kubectl apply -f service.yaml

Verify Deployment

Run the following commands to check your deployment and services:

kubectl get deployments
kubectl get services

Grab the external IP provided and open it in your browser:

Does the response look familiar? That's right, the API we built in the previous post is now accessible via the internet through our Kubernetes cluster! 🎉 (Of course, since we're using a dummy database, it only shows the most recent data entry. 🤷‍♂️)

Feel free to run your tests and deploy some exciting features!

Cleaning Up Resources

Don't forget to clean up all the resources created during this process, including EC2 instances, EBS volumes, load balancers, Elastic IPs, resource groups, ECR repositories, EKS clusters, node groups, CloudFormation stacks, VPCs, subnets, NAT gateways, and network interfaces. Here's an order I recommend:

Node groups
EKS cluster
ECR repositories
Security groups
Load balancers
DHCP option sets
NAT gateways

Although you can use aws-nuke to delete resources easily, beware: if used incorrectly, it can delete critical resources. Therefore, I recommend manually cleaning up the resources.

For a better understanding of the resources created, check out AWS CloudTrail. It will show you every command executed by the stack, providing valuable insights into the automated processes.

What's Next? 🚀

Now that your FastAPI application is successfully deployed on AWS EKS, it's time to think about enhancing and scaling your setup further. Here are some next steps to take your deployment to the next level:

Authentication and Authorization: Implement robust authentication mechanisms like OAuth2 or JWT to secure your APIs.
Improving Autoscaling: Leverage Kubernetes' Horizontal Pod Autoscaler (HPA) to dynamically adjust the number of pod replicas based on real-time metrics like CPU usage or custom application metrics.
Observability: Integrate monitoring and logging solutions like Prometheus and Grafana for metrics, and use tools like Fluentd and AWS CloudWatch for logging to ensure you have full visibility into your application's performance.
High Availability: Design your system for high availability by deploying your application across multiple availability zones and using Kubernetes Ingress controllers for better load balancing and failover capabilities.

In the dynamic world of modern tech, the principles of observability, scalability, and security are paramount, especially when deploying data-centric applications. By implementing these best practices, you're not only future-proofing your infrastructure but also ensuring a seamless experience for end-users.

Moreover, the ability to efficiently deploy machine learning models and data products is becoming increasingly crucial for data scientists and engineers. Leveraging tools like EKS for MLOps (Machine Learning Operations) and LLMOps (Large Language Model Operations) can enable you to deliver robust and scalable AI solutions. Every model deployment can benefit from the high availability, enhanced observability, and seamless scaling these platforms provide. In an era where data is the cornerstone of innovation, mastering these advanced deployment techniques will set you apart.

Stay tuned for future tutorials where we'll dive deeper into these topics, covering advanced Kubernetes features, CI/CD pipelines, and integrating more sophisticated data workflows. Until then, happy deploying! 🎉

Github repo:
https://github.com/felipe-de-godoy/FastAPI-for-Data-Delivery

Credit for Cover image https://dev.to/bravinsimiyu

FastAPI for Data Applications: From Concept to Creation. Part I

Felipe de Godoy — Fri, 14 Jun 2024 00:58:11 +0000

In this blog post, we'll explore how to create an API using FastAPI, a modern Python framework designed for building APIs with high performance. We will create a simple API that allows users to add, update, and query items stored temporarily in memory. Alongside this, we'll discuss how you can extend this example to expose machine learning models, perform online processing in decision engines, and ensure best practices for a robust, secure API.

Pre-requisites: Installation of FastAPI and Uvicorn

Before diving into the code, we need to install FastAPI and Uvicorn, an ASGI server to run our application. Run the following command in your terminal:

pip install fastapi uvicorn

If you prefer, you can run in a virtual environment to isolate your execution.

Project Structure

We'll start by creating a file named main.py, which will contain all the code for our application.

1. Setting Up FastAPI

The first step is to import the necessary modules and initialize the FastAPI app.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI()

FastAPI: The core library for building the API.
HTTPException: This handles HTTP exceptions.
BaseModel: Part of Pydantic, used for data validation.
List: Used for type hinting.

2. Defining the Data Model

Next, we'll define the data model using Pydantic's BaseModel. This model will describe the structure of the data we want to store and process.

class Item(BaseModel):
    id: int
    name: str
    description: str = None

Item: Our data model with three fields: id, name, and description. The description field is optional.

3. In-Memory Database

For simplicity, we'll use a list to simulate a database. Here you can insert the connectors for your specific use case.

fake_db: List[Item] = []

fake_db: This list will store instances of Item during the application's runtime.

4. Define the Root Endpoint

We'll start by defining a simple root endpoint to ensure our application is running.

@app.get("/")
async def root():
    return {"message": "Index Endpoint"}

This endpoint returns a JSON message saying "Index Endpoint" when accessed.

5. Implementing API Endpoints

a) Retrieve All Items

To retrieve all items in our "database", we'll define a GET endpoint.

@app.get("/items/", response_model=List[Item])
async def get_items():
    return fake_db

This endpoint returns the list of all items.

b) Retrieve a Specific Item

To retrieve a specific item by its ID:

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    for item in fake_db:
        if item.id == item_id:
            return item
    raise HTTPException(status_code=404, detail="Item not found")

item_id: The item's ID passed as a path parameter.
This function searches the list for an item with the given ID and returns it if found. If not, it raises a 404 error.

c) Create a New Item

To add new items to our list:

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    for existing_item in fake_db:
        if existing_item.id == item.id:
            raise HTTPException(status_code=400, detail="Item already exists")
    fake_db.append(item)
    return item

create_item: This function checks if an item with the same ID already exists. If it does, it raises a 400 error. Otherwise, it adds the item to the list.

d) Update an Existing Item

To update an existing item by its ID:

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, updated_item: Item):
    for idx, existing_item in enumerate(fake_db):
        if existing_item.id == item_id:
            fake_db[idx] = updated_item
            return updated_item
    raise HTTPException(status_code=404, detail="Item not found")

update_item: The function updates an item if it exists, otherwise raises a 404 error.

Running the Application

Save your file as main.py. To run the application, execute the following command:

uvicorn main:app --reload

The --reload flag enables auto-reloading, meaning the server will restart whenever you make changes to the code. This is useful during development.

Testing Your API

You can test the API using tools like curl, Postman, bash terminal, or even a web browser. Here are some curl commands for testing:

a) Add New Items

curl -X POST "http://127.0.0.1:8000/items/" -H "Content-Type: application/json" -d '{"id": 1, "name": "Item 1", "description": "This is item 1"}'

curl -X POST "http://127.0.0.1:8000/items/" -H "Content-Type: application/json" -d '{"id": 2, "name": "Item 2", "description": "This is item 2"}'

b) Update an Item

curl -X PUT "http://127.0.0.1:8000/items/1" -H "Content-Type: application/json" -d '{"id": 1, "name": "Updated Item 1", "description": "This is the updated item 1"}'

c) Retrieve All Items

curl -X GET "http://127.0.0.1:8000/items/"

d) Retrieve a Specific Item

curl -X GET "http://127.0.0.1:8000/items/1"

Advanced Use Cases

Exposing a Machine Learning Model

One of the most exciting applications of FastAPI is exposing machine learning models. Imagine you want to serve a trained model for real-time predictions. Here’s an example of how to load a machine-learning model and expose it via an endpoint.

Firstly, ensure you have a trained machine-learning model stored in a pickled file. For this example, we'll assume you have a model.pkl file.

import pickle
from sklearn.ensemble import RandomForestClassifier

# Load the model
with open("model.pkl", "rb") as f:
    model = pickle.load(f)

@app.post("/predict/")
async def predict(data: List[float]):
    prediction = model.predict([data])
    return {"prediction": prediction[0]}

Data Validation: Validate input data properly to avoid unexpected errors. Pydantic models can help structure the input data for the prediction endpoint.
Performance: Be mindful of the model's loading time and how often it's accessed. For high-frequency endpoints, consider model optimization and caching mechanisms.
Risks: Make sure only validated and sanitized data is fed into your model to avoid security vulnerabilities like input injection.

Online Processing in a Decision Engine

Another practical scenario is implementing online data processing using an API. For example, in a financial application, you might need to decide based on streaming data, such as approving or declining a transaction.

@app.post("/process/")
async def process_transaction(transaction: dict):
    # Perform some real-time computation/decision
    decision = "approved" if transaction["amount"] < 1000 else "declined"
    return {"transaction_id": transaction["transaction_id"], "decision": decision}

Efficiency: Ensure your processing logic is efficient since it could potentially process numerous transactions in real-time.
Consistency: Maintain the consistency and idempotence of your decision logic to handle scenarios where the same request might be processed multiple times.
Security: Always validate incoming data to protect against malicious payloads.

Conclusion

In this blog post, we've built a simple API using FastAPI that allows you to add, update, and fetch items stored in memory. Additionally, we discussed how you can extend this architecture to expose machine learning models and perform online processing in decision engines.

This example serves as a foundational understanding of how to use FastAPI for basic CRUD operations and more advanced use cases. For more complex scenarios, consider connecting to a real database, implementing robust authentication, and ensuring high availability.

Feel free to explore and expand upon this example to suit your needs. Happy coding!

Looking Ahead to Part 2: Dockerizing and Running in Kubernetes

In the next part of our series, we will take this FastAPI application to the next level by preparing it for production deployment. We will walk you through the process of Dockerizing the application, ensuring it is containerized for consistency and portability. Then, we'll delve into orchestrating the application using Kubernetes, providing the scalability and reliability required for production environments. This includes setting up Kubernetes manifests, deploying the application to a Kubernetes cluster, and managing the application lifecycle in a cloud-native environment. Stay tuned to learn how to transform your development setup into a robust, enterprise-ready deployment! Keep an eye out for Part 2 of this series as we dive into Docker and Kubernetes.

Github repo: https://github.com/felipe-de-godoy/FastAPI-for-Data-Delivery

Credits: Image from "Real Python"

DEV Community: Felipe de Godoy

FastAPI for Data Applications: Deploying with AWS Lambda. Part III

Prerequisites

Step-by-Step Guide to Exposing Data with FastAPI through AWS Lambda

1. Set Up Your FastAPI Application

2. Create a requirements.txt File

3. Create a Dockerfile for the Serverless Framework

4. Serverless Framework Configuration

5. Update serverless.yml Configuration

6. Deploy the Application

7. Verify the Deployment

Conclusion

Aggregation in GROUP BY vs. Window Functions Using OVER()

Understanding Aggregation in GROUP BY

Syntax:

Introducing Window Functions Using OVER()

Syntax:

Structure of Window Functions

1. FUNCTION()

2. OVER()

3. PARTITION BY column1

4. ORDER BY column2

Comparing GROUP BY and Window Functions

Data Reduction vs. Data Preservation

Context and Insight

Business Applications

Practical Performance Tips

Conclusion

SQL #DataEngineering #WindowFunctions #BigData #Programming #Database

Leveraging PySpark.Pandas for Efficient Data Pipelines

Why Use Pandas API on Spark?

Setting Up Your Environment

Getting the Data

Processing Data with Pandas API on Spark

Conclusion

Automating AWS Cost and Usage Report with CloudFormation

Automating AWS Cost and Usage Report with CloudFormation

Prerequisites

Project Structure

Detailed Steps

1. Create the S3 Bucket

2. Write your Cloudformation Template

cloudformation/MasterTemplate.yaml

3. Running the CloudFormation Stack

4. Create an Athena Table

5. Retrieve your Data

athena/example_query.sql

Conclusion

FastAPI for Data Applications: Dockerizing and Scaling Your API on Kubernetes. Part II

Prerequisites

Step 1: Dockerize Your FastAPI Application

The FastAPI Application

Creating the Dockerfile

Create the requirements.txt file

Building and Running the Docker Image Locally

Step 2: Set Up an EKS Cluster

Step 3: Push Docker Image to AWS ECR

Step 4: Deploy to EKS

Verify Deployment

Cleaning Up Resources

What's Next? 🚀

FastAPI for Data Applications: From Concept to Creation. Part I

Pre-requisites: Installation of FastAPI and Uvicorn

Project Structure

1. Setting Up FastAPI

2. Defining the Data Model

3. In-Memory Database

4. Define the Root Endpoint

5. Implementing API Endpoints

a) Retrieve All Items

b) Retrieve a Specific Item

c) Create a New Item

d) Update an Existing Item

Running the Application

Testing Your API

a) Add New Items

b) Update an Item

c) Retrieve All Items

d) Retrieve a Specific Item

Advanced Use Cases

2. Create a `requirements.txt` File

3. Create a `Dockerfile` for the Serverless Framework

5. Update `serverless.yml` Configuration

`cloudformation/MasterTemplate.yaml`

`athena/example_query.sql`

Create the `requirements.txt` file