Batch ETL Pipeline with Glue and pgvector

Build a Terraform-managed ETL pipeline loading vector embeddings into PostgreSQL.

Introduction

30 Second Summary

Every time a streaming service recommends a movie you actually want to watch, there is a data pipeline behind the scenes transforming raw user behavior into searchable patterns. These pipelines are the backbone of modern recommendation systems.

In this project, you will build a batch ETL pipeline that ingests raw movie rating data, transforms it with a serverless PySpark job, and loads vector embeddings into a database for similarity search. All infrastructure is defined and deployed from code using Terraform.

What You'll Build

A production-style batch ETL pipeline on AWS that transforms raw CSV data into queryable vector embeddings using Terraform, AWS Glue, Amazon S3, and Amazon RDS PostgreSQL with pgvector.

By the end of this project, you'll have:

  • A Terraform configuration that provisions S3, RDS PostgreSQL, IAM roles, and a Glue ETL job from code.
  • A PySpark ETL script running on AWS Glue 5.0 that cleans raw data and engineers feature vectors.
  • A pgvector-enabled PostgreSQL database serving cosine similarity queries on movie embeddings.
  • Secret Mission: Add a Glue Crawler to auto-catalog your S3 data and query it with Amazon Athena.

Are there any prerequisites?

You need an AWS account with the AWS CLI configured, Terraform installed, Python 3.10, and VS Code. Step 1 walks you through verifying each tool and installing any missing Python packages.

Before We Start

Building a data pipeline requires several tools working together. Terraform provisions infrastructure, the AWS CLI connects you to your cloud account, and Python runs your local scripts.

Before writing any code, let's verify all tools are ready, create a project structure, and install the Python packages you'll need later.

In this step, get ready to:
  • Verify Terraform, AWS CLI, and Python are working at the correct versions.
  • Create the project directory structure with separate folders for each concern.
  • Set up a Python virtual environment and install required packages.
Verify your tools are installed

Let's confirm each tool is present and running the version this project needs.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then typing Terminal and pressing Enter.
  • Check your Terraform version by running this command:
terraform --version

✔️ I see version 1.5.0 or higher

You're good to go. Any version 1.5.0 or above works for this project.

ⓧ Command not found or version below 1.5.0

Terraform needs to be installed or upgraded. Follow the official installation steps for your OS:

  • On macOS with Homebrew, run:
  • brew tap hashicorp/tap followed by brew install hashicorp/tap/terraform.
  • On Windows or Linux, visit the Terraform install page and follow the instructions for your platform.
  • After installing, run terraform --version again to confirm.
  • Verify your AWS identity by running this command:
aws sts get-caller-identity

✔️ I see my account information

You should see a JSON response containing your Account ID, UserId, and Arn. This confirms your CLI is authenticated.

ⓧ I see an error

This usually means your AWS CLI credentials are not configured. Run:

aws configure

Enter your Access Key ID, Secret Access Key, default region (e.g. us-east-1), and output format (json). Then retry the get-caller-identity command.

  • Confirm your Python version by running this command:
python3 --version

✔️ I see Python 3.10

Python 3.10 is ready. This is the version your local scripts will use.

ⓧ I see a different version or command not found

This project requires Python 3.10. Visit the Python downloads page to install 3.10 for your platform.

On macOS, if you installed via Homebrew, try running python3.10 --version directly. The system default may point to a different version.

What does aws sts get-caller-identity do?

This command asks AWS "who am I?" and returns the account, user, and role currently configured in your CLI. It confirms your credentials are valid without making any changes to your account.

Create the project directory structure

Each part of this pipeline lives in its own folder. Keeping Terraform config, Glue scripts, sample data, and local scripts separate makes the project easier to navigate and maintain.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir etl-pipeline
cd etl-pipeline
  • Create the four subdirectories by running this command:
mkdir terraform glue data scripts
  • Confirm the folders exist by running:
ls

You should see terraform, glue, data, and scripts listed in the output.

What is each folder for?

  • terraform/ holds your infrastructure-as-code configuration (S3, RDS, Glue resources).
  • glue/ holds the PySpark ETL script that runs on AWS Glue.
  • data/ holds the sample movie recommendation dataset.
  • scripts/ holds local Python scripts for loading vectors and running queries.

Not seeing all four folders?

Make sure you ran the mkdir command from inside the etl-pipeline folder. Run pwd to check your current location. It should end with etl-pipeline.

help me figure out why my folders weren't created correctly

Set up your Python virtual environment

A virtual environment isolates this project's Python packages from your system-wide installs. This prevents version conflicts between projects.

  • Create a virtual environment by running this command:
python3 -m venv venv
  • Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

  • Install the required packages by running this command:
pip install psycopg2-binary pgvector boto3 numpy pyarrow

What are these packages for?

  • psycopg2-binary connects Python to your PostgreSQL database.
  • pgvector adds vector data type support to your database queries.
  • boto3 is the AWS SDK for Python, used to interact with S3.
  • numpy handles numerical array operations for building embeddings.
  • pyarrow reads Parquet files locally after downloading from S3.
  • Verify all packages installed correctly by running this command:
python3 -c "import psycopg2, pgvector, boto3, numpy, pyarrow; print('All packages imported successfully')"

✔️ I see the success message

All five packages are installed and importable. Your environment is ready.

ⓧ I see an ImportError

Make sure your virtual environment is active. You should see (venv) in your prompt. If not, re-run the activation command above and then re-run the pip install command.

help me fix my Python package installation error

Your environment is ready and your project structure is in place. Next up, you'll write the Terraform configuration that defines your entire AWS infrastructure as code.

Set Up Project Structure and Sample Data

Every data pipeline starts with data to process. Before writing any Terraform configuration or ETL logic, you need a dataset that represents what your pipeline will ingest and transform.

In this step, you'll create a sample movie recommendations dataset and a placeholder for the AWS Glue script that will process it later.

In this step, get ready to:
  • Create a 50-record movie dataset as a CSV file.
  • Add a placeholder Glue ETL script for future development.
Create the sample movie dataset

The pipeline will read raw movie recommendation data from S3, transform it into feature vectors, and load those vectors into a pgvector database. You need a local CSV file that you'll upload to S3 in the next step.

  • In VS Code, create a new file called movies.csv inside your data/ folder.
  • Paste the following content into movies.csv:
movie_id,title,genre,avg_rating,num_ratings,year
1,The Matrix,SciFi,4.7,25000,1999
2,Inception,SciFi,4.6,22000,2010
3,The Godfather,Drama,4.9,30000,1972
4,Pulp Fiction,Drama,4.5,28000,1994
5,The Dark Knight,Action,4.8,35000,2008
6,Forrest Gump,Drama,4.6,27000,1994
7,Fight Club,Drama,4.5,20000,1999
8,Interstellar,SciFi,4.5,18000,2014
9,The Shawshank Redemption,Drama,4.9,32000,1994
10,Gladiator,Action,4.3,15000,2000
11,The Lion King,Animation,4.4,22000,1994
12,Toy Story,Animation,4.3,19000,1995
13,Jurassic Park,SciFi,4.2,21000,1993
14,Titanic,Romance,4.1,26000,1997
15,Avatar,SciFi,3.9,24000,2009
16,The Avengers,Action,4.1,29000,2012
17,Spider-Man,Action,3.8,16000,2002
18,Finding Nemo,Animation,4.2,17000,2003
19,Up,Animation,4.4,14000,2009
20,WALL-E,Animation,4.3,12000,2008
21,The Prestige,Drama,4.5,13000,2006
22,Memento,Drama,4.4,11000,2000
23,Django Unchained,Action,4.3,18000,2012
24,Mad Max Fury Road,Action,4.4,15000,2015
25,Blade Runner 2049,SciFi,4.1,9000,2017
26,Arrival,SciFi,4.2,10000,2016
27,Parasite,Drama,4.6,14000,2019
28,Coco,Animation,4.5,13000,2017
29,Inside Out,Animation,4.3,15000,2015
30,Dune,SciFi,4.3,12000,2021
31,The Truman Show,Drama,4.2,11000,1998
32,Good Will Hunting,Drama,4.4,12000,1997
33,Schindlers List,Drama,4.8,20000,1993
34,Braveheart,Action,4.2,14000,1995
35,The Terminator,Action,4.1,16000,1984
36,Aliens,SciFi,4.3,15000,1986
37,Back to the Future,SciFi,4.5,18000,1985
38,E.T.,SciFi,4.1,17000,1982
39,Rocky,Drama,4.0,13000,1976
40,Star Wars,SciFi,4.7,35000,1977
41,The Silence of the Lambs,Drama,4.5,16000,1991
42,Saving Private Ryan,Action,4.5,19000,1998
43,The Green Mile,Drama,4.6,14000,1999
44,Se7en,Drama,4.4,15000,1995
45,Whiplash,Drama,4.5,10000,2014
46,La La Land,Romance,4.1,12000,2016
47,Spirited Away,Animation,4.6,11000,2001
48,Your Name,Animation,4.4,8000,2016
49,John Wick,Action,4.0,14000,2014
50,Wonder Woman,Action,3.9,13000,2017

What does this data represent?

Each row is a movie with six attributes that the ETL pipeline will use for feature engineering.

  • movie_id is the unique identifier for each record.
  • avg_rating and num_ratings are the numeric features that get normalized into vector dimensions.
  • genre becomes a one-hot encoded dimension via PySpark's StringIndexer.
  • year gets normalized to capture temporal similarity between movies.
  • Save data/movies.csv.
  • Confirm the file exists by running:
ls data/

You should see movies.csv listed in the output.

Don't see movies.csv?

Make sure you created the file inside the data/ subfolder, not in the project root. Check for typos in the filename.

help me find where my movies.csv file ended up

✔️ Awesome, I've got everything!

Great. Double check you've saved the file before moving on.

ⓧ I'd like to double check the full code

movie_id,title,genre,avg_rating,num_ratings,year
1,The Matrix,SciFi,4.7,25000,1999
2,Inception,SciFi,4.6,22000,2010
3,The Godfather,Drama,4.9,30000,1972
4,Pulp Fiction,Drama,4.5,28000,1994
5,The Dark Knight,Action,4.8,35000,2008
6,Forrest Gump,Drama,4.6,27000,1994
7,Fight Club,Drama,4.5,20000,1999
8,Interstellar,SciFi,4.5,18000,2014
9,The Shawshank Redemption,Drama,4.9,32000,1994
10,Gladiator,Action,4.3,15000,2000
11,The Lion King,Animation,4.4,22000,1994
12,Toy Story,Animation,4.3,19000,1995
13,Jurassic Park,SciFi,4.2,21000,1993
14,Titanic,Romance,4.1,26000,1997
15,Avatar,SciFi,3.9,24000,2009
16,The Avengers,Action,4.1,29000,2012
17,Spider-Man,Action,3.8,16000,2002
18,Finding Nemo,Animation,4.2,17000,2003
19,Up,Animation,4.4,14000,2009
20,WALL-E,Animation,4.3,12000,2008
21,The Prestige,Drama,4.5,13000,2006
22,Memento,Drama,4.4,11000,2000
23,Django Unchained,Action,4.3,18000,2012
24,Mad Max Fury Road,Action,4.4,15000,2015
25,Blade Runner 2049,SciFi,4.1,9000,2017
26,Arrival,SciFi,4.2,10000,2016
27,Parasite,Drama,4.6,14000,2019
28,Coco,Animation,4.5,13000,2017
29,Inside Out,Animation,4.3,15000,2015
30,Dune,SciFi,4.3,12000,2021
31,The Truman Show,Drama,4.2,11000,1998
32,Good Will Hunting,Drama,4.4,12000,1997
33,Schindlers List,Drama,4.8,20000,1993
34,Braveheart,Action,4.2,14000,1995
35,The Terminator,Action,4.1,16000,1984
36,Aliens,SciFi,4.3,15000,1986
37,Back to the Future,SciFi,4.5,18000,1985
38,E.T.,SciFi,4.1,17000,1982
39,Rocky,Drama,4.0,13000,1976
40,Star Wars,SciFi,4.7,35000,1977
41,The Silence of the Lambs,Drama,4.5,16000,1991
42,Saving Private Ryan,Action,4.5,19000,1998
43,The Green Mile,Drama,4.6,14000,1999
44,Se7en,Drama,4.4,15000,1995
45,Whiplash,Drama,4.5,10000,2014
46,La La Land,Romance,4.1,12000,2016
47,Spirited Away,Animation,4.6,11000,2001
48,Your Name,Animation,4.4,8000,2016
49,John Wick,Action,4.0,14000,2014
50,Wonder Woman,Action,3.9,13000,2017
Create a placeholder Glue script

The Terraform configuration you'll write in the next step references a Glue script file at glue/etl_job.py. You'll write the full PySpark logic in Step 4, but the file needs to exist now so Terraform can upload it to S3.

  • Create an empty placeholder script by running:
touch glue/etl_job.py
  • Confirm both project files exist by running:
ls data/ glue/

You should see movies.csv under data/ and etl_job.py under glue/.

File not showing up?

Make sure your terminal is in the project root directory (the one containing the terraform/, glue/, data/, and scripts/ folders).

help me figure out why my project files aren't showing up in the right folders

Your sample data and project structure are ready. Next up, you'll write the Terraform configuration that provisions your entire AWS infrastructure from code.

Write Terraform Infrastructure Configuration

With your project structure and sample data in place, the next step is to define the cloud infrastructure that will power your ETL pipeline. Terraform lets you describe all of these resources in code, so you can create, update, and destroy them repeatably.

In this step, you will write three Terraform configuration files that define an S3 bucket for data storage, an RDS PostgreSQL database for vector embeddings, and an AWS Glue job for running the ETL transformation.

In this step, get ready to:
  • Define the Terraform provider and S3 data lake resources.
  • Configure an RDS PostgreSQL instance with a security group.
  • Create the Glue IAM role and ETL job resource.
Create the provider block and S3 resources

Every Terraform project starts with a variables file that defines the inputs your configuration needs. These variables let you reuse the same configuration across environments without hardcoding values.

  • In VS Code, create a new file called variables.tf inside your terraform/ folder.
  • Paste the following code into variables.tf:
variable "aws_region" {
  description = "AWS region to deploy resources"
  type        = string
  default     = "us-east-1"
}

variable "bucket_name" {
  description = "S3 bucket name for the data lake"
  type        = string
}

variable "db_username" {
  description = "RDS master username"
  type        = string
  default     = "postgres"
}

variable "db_password" {
  description = "RDS master password"
  type        = string
  sensitive   = true
}

variable "my_ip_cidr" {
  description = "Your IP address in CIDR notation for RDS access (e.g. 203.0.113.1/32)"
  type        = string
}

What does this code do?

  • aws_region sets the AWS region to deploy into, defaulting to us-east-1.
  • bucket_name is the globally unique name you will choose for your S3 data lake bucket.
  • db_password is marked sensitive so Terraform never prints it in plan output or logs.
  • my_ip_cidr restricts database access to your IP address only, following the principle of least privilege.
  • Save variables.tf.

Now create the main configuration file that defines the S3 bucket and data upload resources.

  • Create a new file called main.tf inside the terraform/ folder.
  • Paste the following code into main.tf:
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 6.0"
    }
  }
  required_version = ">= 1.5.0"
}

provider "aws" {
  region = var.aws_region
}

resource "aws_s3_bucket" "data_lake" {
  bucket        = var.bucket_name
  force_destroy = true
}

resource "aws_s3_object" "raw_data" {
  bucket = aws_s3_bucket.data_lake.id
  key    = "raw/movies.csv"
  source = "../data/movies.csv"
  etag   = filemd5("../data/movies.csv")
}

resource "aws_s3_object" "glue_script" {
  bucket = aws_s3_bucket.data_lake.id
  key    = "scripts/etl_job.py"
  source = "../glue/etl_job.py"
  etag   = filemd5("../glue/etl_job.py")
}

What does this code do?

  • The terraform block pins the AWS provider to version ~> 6.0 so your configuration stays compatible with the latest 6.x releases.
  • aws_s3_bucket creates the data lake bucket with force_destroy = true so Terraform can delete the bucket even if it still contains objects (handy for cleanup).
  • The two aws_s3_object resources upload your local CSV and Glue script to S3. The etag field uses filemd5() so Terraform re-uploads the file whenever it changes.
  • Save main.tf.

Seeing a red squiggly in VS Code?

If your editor underlines var.aws_region or var.bucket_name, that is fine. The HashiCorp Terraform extension may not resolve variables until you run terraform init. These warnings disappear once the provider is installed.

help me fix Terraform variable reference warnings in VS Code

Create the RDS PostgreSQL instance

Your pipeline needs a database to store the final vector embeddings. You will define a security group that restricts access to your IP, then create a PostgreSQL 16 instance with the pgvector extension available.

  • In main.tf, add the following code below the aws_s3_object resources you just wrote:
resource "aws_security_group" "rds_sg" {
  name        = "rds-recommendation-sg"
  description = "Allow PostgreSQL access"

  ingress {
    from_port   = 5432
    to_port     = 5432
    protocol    = "tcp"
    cidr_blocks = [var.my_ip_cidr]
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_db_instance" "postgres" {
  identifier             = "recommendation-db"
  engine                 = "postgres"
  engine_version         = "16"
  instance_class         = "db.t3.micro"
  allocated_storage      = 20
  storage_type           = "gp2"
  db_name                = "recommendations"
  username               = var.db_username
  password               = var.db_password
  publicly_accessible    = true
  skip_final_snapshot    = true
  vpc_security_group_ids = [aws_security_group.rds_sg.id]
}

What does this code do?

  • The security group opens port 5432 (PostgreSQL's default) only to the CIDR you supply in my_ip_cidr. All other inbound traffic is blocked.
  • db.t3.micro is the smallest general-purpose RDS instance. It is free-tier eligible for accounts under 12 months old.
  • skip_final_snapshot = true lets Terraform destroy the database without forcing a snapshot, making cleanup faster.
  • publicly_accessible = true allows your local machine to connect directly. The security group still restricts access to your IP.

Why make RDS publicly accessible?

A private RDS instance would require a VPC with private subnets, a NAT Gateway (about $0.045/hr), and a Glue VPC Connection. Making the instance public and restricting access via the security group eliminates that complexity and cost while still limiting connections to your IP only.

  • Save main.tf.

Syntax error on the security group block?

Make sure each ingress and egress block has its own pair of curly braces nested inside the aws_security_group block. HCL is sensitive to brace placement.

help me debug a syntax error in my Terraform security group resource

Create the Glue IAM role and Glue job resource

AWS Glue needs an IAM role that grants it permission to run jobs and access your S3 bucket. You will attach the AWS-managed AWSGlueServiceRole policy for basic Glue operations, then add a custom inline policy for S3 read/write access.

  • In main.tf, add the following code below the aws_db_instance resource:
resource "aws_iam_role" "glue_role" {
  name = "glue-etl-recommendation-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "glue.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "glue_service" {
  role       = aws_iam_role.glue_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
}

resource "aws_iam_role_policy" "glue_s3_access" {
  name = "glue-s3-access"
  role = aws_iam_role.glue_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:GetObject",
          "s3:PutObject",
          "s3:ListBucket"
        ]
        Resource = [
          aws_s3_bucket.data_lake.arn,
          "${aws_s3_bucket.data_lake.arn}/*"
        ]
      }
    ]
  })
}

What does this code do?

  • The assume_role_policy tells AWS that only the Glue service (glue.amazonaws.com) can assume this role.
  • AWSGlueServiceRole is an AWS-managed policy giving Glue the baseline permissions it needs (CloudWatch logs, catalog access).
  • The custom inline policy restricts S3 access to only your data lake bucket. This follows least-privilege: Glue can read input and write output, but nothing else.
  • Save main.tf.

Now define the Glue job itself, which tells AWS how to run your ETL script.

  • Still in main.tf, add the following code below the aws_iam_role_policy resource:
resource "aws_glue_job" "etl_job" {
  name              = "recommendation-etl-job"
  role_arn          = aws_iam_role.glue_role.arn
  glue_version      = "5.0"
  number_of_workers = 2
  worker_type       = "G.1X"
  timeout           = 10
  max_retries       = 0

  command {
    script_location = "s3://${aws_s3_bucket.data_lake.bucket}/scripts/etl_job.py"
    name            = "glueetl"
    python_version  = "3"
  }

  default_arguments = {
    "--job-language"                     = "python"
    "--enable-continuous-cloudwatch-log" = "true"
    "--enable-metrics"                   = ""
    "--S3_BUCKET"                        = aws_s3_bucket.data_lake.bucket
  }
}

What does this code do?

  • glue_version = "5.0" runs the job on the latest Glue runtime (Spark 3.5.4, Python 3.11). This gives the best performance at the lowest cost per job.
  • worker_type = "G.1X" with 2 workers is the minimum for a PySpark batch job. Each G.1X worker has 4 vCPU and 16 GB memory.
  • timeout = 10 sets a 10-minute maximum. The default is 2880 minutes (48 hours). A tight timeout prevents accidental runaway charges if the script has a bug.
  • The --S3_BUCKET default argument passes the bucket name to your PySpark script at runtime so it knows where to read and write data.
  • Save main.tf.

Finally, create an outputs file so Terraform prints the key resource identifiers after deployment.

  • Create a new file called outputs.tf inside the terraform/ folder.
  • Paste the following code into outputs.tf:
output "s3_bucket_name" {
  value = aws_s3_bucket.data_lake.bucket
}

output "rds_endpoint" {
  value = aws_db_instance.postgres.endpoint
}

output "glue_job_name" {
  value = aws_glue_job.etl_job.name
}

What does this code do?

Outputs print values to your terminal after terraform apply completes. You will use rds_endpoint to connect your local script to the database, and glue_job_name to start the ETL job from the CLI.

  • Save outputs.tf.

✔️ Awesome, I've got everything!

Great. Make sure you have saved all three files: variables.tf, main.tf, and outputs.tf.

ⓧ I'd like to double check the full code

Here is the full terraform/main.tf:

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 6.0"
    }
  }
  required_version = ">= 1.5.0"
}

provider "aws" {
  region = var.aws_region
}

resource "aws_s3_bucket" "data_lake" {
  bucket        = var.bucket_name
  force_destroy = true
}

resource "aws_s3_object" "raw_data" {
  bucket = aws_s3_bucket.data_lake.id
  key    = "raw/movies.csv"
  source = "../data/movies.csv"
  etag   = filemd5("../data/movies.csv")
}

resource "aws_s3_object" "glue_script" {
  bucket = aws_s3_bucket.data_lake.id
  key    = "scripts/etl_job.py"
  source = "../glue/etl_job.py"
  etag   = filemd5("../glue/etl_job.py")
}

resource "aws_security_group" "rds_sg" {
  name        = "rds-recommendation-sg"
  description = "Allow PostgreSQL access"

  ingress {
    from_port   = 5432
    to_port     = 5432
    protocol    = "tcp"
    cidr_blocks = [var.my_ip_cidr]
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_db_instance" "postgres" {
  identifier             = "recommendation-db"
  engine                 = "postgres"
  engine_version         = "16"
  instance_class         = "db.t3.micro"
  allocated_storage      = 20
  storage_type           = "gp2"
  db_name                = "recommendations"
  username               = var.db_username
  password               = var.db_password
  publicly_accessible    = true
  skip_final_snapshot    = true
  vpc_security_group_ids = [aws_security_group.rds_sg.id]
}

resource "aws_iam_role" "glue_role" {
  name = "glue-etl-recommendation-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "glue.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "glue_service" {
  role       = aws_iam_role.glue_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
}

resource "aws_iam_role_policy" "glue_s3_access" {
  name = "glue-s3-access"
  role = aws_iam_role.glue_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:GetObject",
          "s3:PutObject",
          "s3:ListBucket"
        ]
        Resource = [
          aws_s3_bucket.data_lake.arn,
          "${aws_s3_bucket.data_lake.arn}/*"
        ]
      }
    ]
  })
}

resource "aws_glue_job" "etl_job" {
  name              = "recommendation-etl-job"
  role_arn          = aws_iam_role.glue_role.arn
  glue_version      = "5.0"
  number_of_workers = 2
  worker_type       = "G.1X"
  timeout           = 10
  max_retries       = 0

  command {
    script_location = "s3://${aws_s3_bucket.data_lake.bucket}/scripts/etl_job.py"
    name            = "glueetl"
    python_version  = "3"
  }

  default_arguments = {
    "--job-language"                     = "python"
    "--enable-continuous-cloudwatch-log" = "true"
    "--enable-metrics"                   = ""
    "--S3_BUCKET"                        = aws_s3_bucket.data_lake.bucket
  }
}

Here is the full terraform/variables.tf:

variable "aws_region" {
  description = "AWS region to deploy resources"
  type        = string
  default     = "us-east-1"
}

variable "bucket_name" {
  description = "S3 bucket name for the data lake"
  type        = string
}

variable "db_username" {
  description = "RDS master username"
  type        = string
  default     = "postgres"
}

variable "db_password" {
  description = "RDS master password"
  type        = string
  sensitive   = true
}

variable "my_ip_cidr" {
  description = "Your IP address in CIDR notation for RDS access (e.g. 203.0.113.1/32)"
  type        = string
}

Here is the full terraform/outputs.tf:

output "s3_bucket_name" {
  value = aws_s3_bucket.data_lake.bucket
}

output "rds_endpoint" {
  value = aws_db_instance.postgres.endpoint
}

output "glue_job_name" {
  value = aws_glue_job.etl_job.name
}
  • Validate your configuration by navigating to the terraform/ directory and running:
cd terraform
terraform validate
  • You should see: Success! The configuration is valid.

Seeing an error from terraform validate?

  • If you see a missing argument error, double-check that every var. reference in main.tf has a matching variable block in variables.tf.
  • If you see an invalid reference error, confirm that resource names like aws_s3_bucket.data_lake match exactly between where they are defined and where they are referenced.
  • Make sure you ran terraform validate from inside the terraform/ folder, not the project root.

help me debug a terraform validate error in my ETL pipeline project

Your infrastructure is fully defined in code. In the next step, you will deploy all of these resources to AWS with a single command.

Deploy Infrastructure with Terraform

Your Terraform configuration is written and validated. Now it is time to bring all those resources to life in your AWS account.

In this step, you will initialize Terraform, preview the deployment plan, and then provision everything with a single command. Once the infrastructure is running, you will connect to your new PostgreSQL database and prepare it for vector embeddings.

In this step, get ready to:
  • Initialize Terraform and preview the deployment plan.
  • Provision all AWS resources with terraform apply.
  • Connect to the RDS database and set up pgvector.
Initialize Terraform and preview the plan

Before Terraform can create resources, it needs to download the AWS provider plugin. It also needs values for the variables you defined (bucket name, database password, and your IP address).

  • Find your public IP address by running this command:
curl -4 ifconfig.me
  • Copy the IP address from the output. You will need it in the next step.
  • Create a terraform.tfvars file inside your terraform folder to supply your variable values. Add the following content:
bucket_name = "[[BUCKET_NAME="yourname-recommendation-data-lake"]]"
db_password = "your-secure-password-here"
my_ip_cidr  = "[[MY_IP_CIDR="203.0.113.1/32"]]"

What does this file do?

Terraform reads terraform.tfvars automatically to fill in any variables that do not have defaults. Each line maps a variable name to your chosen value.

  • bucket_name must be globally unique across all AWS accounts. Include your name or a random suffix.
  • db_password is the master password for your PostgreSQL database. Choose something at least 8 characters long.
  • my_ip_cidr restricts database access to your IP only. Append /32 to your IP address to make it a single-host CIDR block.
  • Save terraform.tfvars.

Not sure what to put for bucket_name?

S3 bucket names must be globally unique, lowercase, and between 3-63 characters. Try something like janedoe-recommendation-data-lake with your own name.

Help me pick a valid S3 bucket name

  • Make sure your terminal is inside the terraform folder, then initialize Terraform by running this command:
terraform init

What does terraform init do?

This command downloads the AWS provider plugin (version ~> 6.0 as specified in your config) and sets up Terraform's working directory. You only need to run it once per project, or after changing providers.

  • You should see a message ending with Terraform has been successfully initialized! in green text.

Seeing a provider installation error?

Check that your internet connection is active. Terraform needs to download the provider from the HashiCorp registry.

If you see a version constraint error, confirm your main.tf has version = "~> 6.0" in the required_providers block.

Help me debug this terraform init error

  • Preview what Terraform will create by running:
terraform plan

What does the plan output show?

The plan lists every resource Terraform will create, marked with a + symbol. You should see entries for your S3 bucket, RDS instance, security group, IAM role, and Glue job.

At the bottom, a summary line shows the total count. Expect something like Plan: 8 to add, 0 to change, 0 to destroy (the exact count depends on your configuration).

  • Confirm the plan output shows resources for the S3 bucket, RDS instance, security group, IAM role, policy attachments, and Glue job.

Seeing variable errors in the plan?

If Terraform prompts you for a variable value, your terraform.tfvars file is not being read. Make sure it is in the same directory where you are running terraform plan (the terraform folder).

Help me fix my terraform.tfvars issue

Provision infrastructure with terraform apply

The plan looks good. Now you will tell Terraform to create all the resources for real in your AWS account.

  • Deploy all resources by running:
terraform apply -auto-approve

Why auto-approve?

The -auto-approve flag skips the interactive yes/no confirmation. Since you just reviewed the plan, you already know what Terraform will create.

The RDS instance takes 5-8 minutes to create. This is normal. AWS is setting up your PostgreSQL database, configuring networking, and running initialization scripts behind the scenes.

  • Wait for the command to finish. You will see a final message that says Apply complete! followed by the output values.
  • View your deployed resource details by running:
terraform output

What do these outputs tell you?

Terraform prints three values you defined in outputs.tf:

  • s3_bucket_name is your data lake bucket where raw CSV and Glue scripts live.
  • rds_endpoint is the hostname and port for your PostgreSQL database (format: hostname:5432).
  • glue_job_name is the name you will use to start the ETL job later.
  • Copy the rds_endpoint value from the output. You will need the hostname part (everything before the colon) to connect to the database.

Apply failed with a permissions error?

Your AWS CLI user needs permissions to create S3 buckets, RDS instances, IAM roles, security groups, and Glue jobs. If you are using a restricted IAM user, ask your account administrator for these permissions.

If the error mentions the S3 bucket name already exists, choose a different bucket_name in your terraform.tfvars and run terraform apply -auto-approve again.

Help me debug my terraform apply error

Connect to RDS and enable pgvector

Your PostgreSQL database is running, but it does not have the vector extension yet. You need to enable pgvector and create the table that will store your movie embeddings.

  • Connect to your RDS database using psql by running this command (replace the hostname with the value from your rds_endpoint output, without the :5432 port suffix):
psql -h [[RDS_HOST="your-rds-endpoint-hostname"]] -p 5432 -U postgres -d recommendations
  • When prompted, enter the db_password you set in terraform.tfvars.
  • You should see the recommendations=> prompt, confirming you are connected to the database.

No psql installed?

If you do not have psql installed, you can install it via brew install libpq (macOS) or download the PostgreSQL client tools for your platform. Alternatively, you will run these same SQL commands through the Python script in a later step.

Connection timed out?

This usually means your IP address in my_ip_cidr does not match your current public IP. Run curl -4 ifconfig.me again to check. If your IP changed, update terraform.tfvars and run terraform apply -auto-approve to update the security group.

Help me debug my RDS connection issue

  • Enable the pgvector extension and create the movies table by running this SQL:
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE IF NOT EXISTS movies (
    movie_id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    genre TEXT,
    avg_rating REAL,
    num_ratings INTEGER,
    year INTEGER,
    embedding vector(4)
);

What does this SQL do?

CREATE EXTENSION vector loads the pgvector extension into your database. This adds the vector data type and distance operators (like <=> for cosine distance) that power similarity search.

The movies table stores both the original movie metadata and a 4-dimensional embedding vector. Later, the ETL pipeline will fill this table with feature vectors computed from the rating data.

  • Verify the extension is active by running:
SELECT extversion FROM pg_extension WHERE extname = 'vector';
  • You should see a version number like 0.8.0 or higher in the output.
  • Confirm the table was created by running:
\d movies
  • You should see the column list including movie_id, title, genre, avg_rating, num_ratings, year, and embedding with type vector(4).
  • Exit the psql session by typing:
\q

Permission denied creating the extension?

The pgvector extension requires the master user (postgres in your config) to install. Make sure you connected with -U postgres and not a different username.

Help me fix my pgvector permission error

Your infrastructure is fully deployed and your database is ready for vector embeddings. Next up, you will write the PySpark ETL script that transforms the raw movie data into feature vectors.

Build the PySpark ETL Script

Your infrastructure is live, but the Glue job is pointing to a placeholder script. It needs real transformation logic to read the raw CSV, clean it, and generate the numeric features that become vector embeddings.

In this step, you will write the full PySpark ETL script that runs inside AWS Glue. The script handles data cleaning, feature engineering, and outputs processed Parquet files to S3.

In this step, get ready to:
  • Write a PySpark script that reads raw CSV data from S3 and cleans it.
  • Add feature engineering that computes normalized vectors for each movie.
  • Upload the completed script to S3 so the Glue job can execute it.
Write the Glue script imports and CSV reader

Every Glue PySpark job starts by initializing a GlueContext and reading its arguments. The --S3_BUCKET argument you defined in the Terraform configuration tells the script which bucket to read from.

  • Open glue/etl_job.py in VS Code. Replace the placeholder content with the following code:
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql import functions as F
from pyspark.sql.types import FloatType
from pyspark.ml.feature import StringIndexer

args = getResolvedOptions(sys.argv, ["JOB_NAME", "S3_BUCKET"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

bucket = args["S3_BUCKET"]

# Read raw CSV from S3
df = spark.read.option("header", "true").csv(f"s3://{bucket}/raw/movies.csv")

What does this code do?

getResolvedOptions pulls the JOB_NAME and S3_BUCKET arguments that Glue passes to the script at runtime. These come from the default_arguments block in your Terraform configuration.

GlueContext wraps Spark with Glue-specific features like the Data Catalog integration. The spark.read.csv() call loads your raw movie data from S3 into a Spark DataFrame for transformation.

Add data cleaning logic

Raw CSV data comes in as strings. The cleaning step casts columns to their proper types, removes incomplete rows, and filters out invalid rating values.

  • Add the following code below the CSV read line in glue/etl_job.py:
# Clean data: cast types, drop nulls, filter invalid ratings
df_clean = (
    df.withColumn("avg_rating", F.col("avg_rating").cast(FloatType()))
    .withColumn("num_ratings", F.col("num_ratings").cast("int"))
    .withColumn("year", F.col("year").cast("int"))
    .dropna(subset=["movie_id", "title", "avg_rating", "num_ratings", "year"])
    .filter((F.col("avg_rating") >= 0) & (F.col("avg_rating") <= 5))
)

What does this code do?

Each withColumn call casts a string column to a numeric type. FloatType() is used for ratings (decimal values), while "int" handles whole numbers.

The dropna call removes any row missing a critical field. The filter keeps only rows where the average rating falls between 0 and 5, discarding corrupted entries.

  • Save glue/etl_job.py.

You won't see a visible change yet. The next chunk adds the feature engineering that produces the output vectors.

Seeing a syntax error in your editor?

Make sure your editor is not trying to resolve the awsglue imports locally. These packages only exist inside the Glue runtime environment. Red underlines on import lines are expected and safe to ignore.

help me understand why my Glue script shows import errors locally

Add feature engineering and Parquet output

Feature engineering transforms raw numbers into normalized values that are comparable across different scales. These normalized features become the dimensions of your vector embeddings.

  • Add the following feature engineering code below the df_clean block in glue/etl_job.py:
# Feature engineering: normalize numeric columns
rating_stats = df_clean.agg(
    F.min("avg_rating").alias("min_r"),
    F.max("avg_rating").alias("max_r"),
    F.min("year").alias("min_y"),
    F.max("year").alias("max_y"),
    F.max("num_ratings").alias("max_n")
).collect()[0]

df_features = df_clean.withColumn(
    "norm_rating",
    (F.col("avg_rating") - F.lit(rating_stats["min_r"]))
    / (F.lit(rating_stats["max_r"]) - F.lit(rating_stats["min_r"]) + 1e-8)
).withColumn(
    "norm_year",
    (F.col("year") - F.lit(rating_stats["min_y"]))
    / (F.lit(rating_stats["max_y"]) - F.lit(rating_stats["min_y"]) + 1e-8)
).withColumn(
    "log_num_ratings",
    F.log1p(F.col("num_ratings")) / F.log1p(F.lit(rating_stats["max_n"]))
)

What does this code do?

The rating_stats aggregation collects the min/max values needed for normalization in a single pass over the DataFrame. .collect()[0] pulls the result to the driver as a Row object.

Each withColumn computes a normalized value between 0 and 1. norm_rating and norm_year use min-max scaling. log_num_ratings uses log scaling because popularity values span a wide range (8,000 to 35,000 in your data).

The 1e-8 prevents division by zero if all values happen to be identical.

  • Add the genre indexing, column selection, and Parquet output below the df_features block:
# Genre indexing for embedding dimension
indexer = StringIndexer(inputCol="genre", outputCol="genre_index")
df_indexed = indexer.fit(df_features).transform(df_features)

# Select output columns
df_output = df_indexed.select(
    "movie_id",
    "title",
    "genre",
    "avg_rating",
    "num_ratings",
    "year",
    "norm_rating",
    "norm_year",
    "log_num_ratings",
    "genre_index"
)

# Write processed data as Parquet
df_output.write.mode("overwrite").parquet(f"s3://{bucket}/processed/movies/")

job.commit()

What does this code do?

StringIndexer converts the text-based genre column into a numeric index. This gives each genre (SciFi, Drama, Action, etc.) a unique number that becomes the fourth dimension of the embedding vector.

The select call keeps only the columns needed for the output. Writing as Parquet (a columnar format) to the processed/movies/ prefix creates the processed zone of your data lake. Finally, job.commit() signals Glue that the job finished successfully.

  • Save glue/etl_job.py.

Your ETL script is complete. It will read raw CSV, clean it, compute four normalized features, and write Parquet output to S3.

Not sure your indentation is correct?

Python is whitespace-sensitive. Make sure the method chains (.withColumn, .dropna, .filter) are indented consistently with 4 spaces. If you see IndentationError later when running, check for mixed tabs and spaces.

help me fix indentation issues in my PySpark Glue script

✔️ Awesome, I've got everything!

Great. Make sure you have saved glue/etl_job.py with all four sections: imports/init, cleaning, feature engineering, and Parquet output.

ⓧ I'd like to double check the full code

Here is the complete glue/etl_job.py file. Compare yours against this:

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql import functions as F
from pyspark.sql.types import FloatType
from pyspark.ml.feature import StringIndexer

args = getResolvedOptions(sys.argv, ["JOB_NAME", "S3_BUCKET"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

bucket = args["S3_BUCKET"]

# Read raw CSV from S3
df = spark.read.option("header", "true").csv(f"s3://{bucket}/raw/movies.csv")

# Clean data: cast types, drop nulls, filter invalid ratings
df_clean = (
    df.withColumn("avg_rating", F.col("avg_rating").cast(FloatType()))
    .withColumn("num_ratings", F.col("num_ratings").cast("int"))
    .withColumn("year", F.col("year").cast("int"))
    .dropna(subset=["movie_id", "title", "avg_rating", "num_ratings", "year"])
    .filter((F.col("avg_rating") >= 0) & (F.col("avg_rating") <= 5))
)

# Feature engineering: normalize numeric columns
rating_stats = df_clean.agg(
    F.min("avg_rating").alias("min_r"),
    F.max("avg_rating").alias("max_r"),
    F.min("year").alias("min_y"),
    F.max("year").alias("max_y"),
    F.max("num_ratings").alias("max_n")
).collect()[0]

df_features = df_clean.withColumn(
    "norm_rating",
    (F.col("avg_rating") - F.lit(rating_stats["min_r"]))
    / (F.lit(rating_stats["max_r"]) - F.lit(rating_stats["min_r"]) + 1e-8)
).withColumn(
    "norm_year",
    (F.col("year") - F.lit(rating_stats["min_y"]))
    / (F.lit(rating_stats["max_y"]) - F.lit(rating_stats["min_y"]) + 1e-8)
).withColumn(
    "log_num_ratings",
    F.log1p(F.col("num_ratings")) / F.log1p(F.lit(rating_stats["max_n"]))
)

# Genre indexing for embedding dimension
indexer = StringIndexer(inputCol="genre", outputCol="genre_index")
df_indexed = indexer.fit(df_features).transform(df_features)

# Select output columns
df_output = df_indexed.select(
    "movie_id",
    "title",
    "genre",
    "avg_rating",
    "num_ratings",
    "year",
    "norm_rating",
    "norm_year",
    "log_num_ratings",
    "genre_index"
)

# Write processed data as Parquet
df_output.write.mode("overwrite").parquet(f"s3://{bucket}/processed/movies/")

job.commit()
Upload the completed script to S3

Your Terraform configuration already manages the S3 upload for glue/etl_job.py via the aws_s3_object resource. Because it uses filemd5() as the etag, Terraform detects that the local file changed and re-uploads it automatically.

  • Switch to the terraform/ directory in your terminal by running:
cd terraform
  • Re-apply the Terraform configuration to upload the updated script by running:
terraform apply -auto-approve

What does this command do?

Terraform compares the current filemd5 hash of glue/etl_job.py against what it stored in state during the last apply. Since you replaced the placeholder with real code, the hash differs. Terraform updates the aws_s3_object.glue_script resource by uploading the new file contents to S3.

  • Confirm the output shows 1 resource changed (the S3 object).
  • Verify the script exists at the expected S3 key by running:
aws s3 ls s3://$(terraform output -raw s3_bucket_name)/scripts/

You should see etl_job.py listed with today's date and a file size larger than the placeholder (approximately 1.5 KB).

Seeing 0 resources changed?

If Terraform reports no changes, it means the file on disk still matches the old hash. Make sure you saved glue/etl_job.py with the complete ETL code before running terraform apply. Re-save the file and run terraform apply -auto-approve again.

help me figure out why Terraform is not detecting my file change

Your ETL script is written and deployed to S3. Next up, you will trigger the Glue job and watch it transform your raw movie data into processed Parquet files with vector-ready features.

Run the Pipeline and Validate Results

Your infrastructure is deployed and the ETL script is staged in S3. Now it is time to trigger the pipeline and confirm that data flows from raw CSV all the way through to queryable vector embeddings in PostgreSQL.

In this step you will kick off the AWS Glue job, download the processed output, load vector embeddings into your pgvector-enabled database, and prove the pipeline works by running a cosine similarity query.

In this step, get ready to:
  • Start and monitor the Glue ETL job until it succeeds.
  • Download the processed Parquet data and write a local vector-loading script.
  • Run a cosine similarity query to validate end-to-end results.
Start and monitor the Glue ETL job

AWS Glue jobs run on fully managed Spark infrastructure. You trigger a run from the CLI, then poll for its completion status.

  • Start the Glue job by running this command:
aws glue start-job-run --job-name recommendation-etl-job

What does this command do?

This tells AWS Glue to execute the PySpark script you uploaded to S3. The command returns a JSON response containing a JobRunId that uniquely identifies this run.

Copy the JobRunId value from the output. You will need it to check the job's progress.

  • You should see output like this:
{
    "JobRunId": "jr_abc123def456..."
}
  • Check the job status by running this command (replace the run ID with your actual value):
aws glue get-job-run --job-name recommendation-etl-job --run-id [[JOB_RUN_ID="your JobRunId from the previous command"]] --query "JobRun.JobRunState" --output text

What should I see?

The command returns a single word representing the current state. You will likely see RUNNING or STARTING at first. Wait 2-3 minutes and run the command again. The job should complete within 3-5 minutes for 50 records.

Keep re-running until you see SUCCEEDED.

  • Confirm the status shows SUCCEEDED before continuing.

Job status shows FAILED?

Check that the IAM role has S3 permissions for your bucket. The most common cause is a mismatch between the bucket name in Terraform and the one the Glue script receives via the --S3_BUCKET argument.

You can view the full error by removing the --query and --output flags from the get-job-run command to see the complete JSON response, including the ErrorMessage field.

Help me debug my failed Glue job run

Download processed data and write the vector loading script

The Glue job wrote processed Parquet files to the processed/movies/ prefix in your S3 bucket. You need to download them locally, then write a Python script that reads the Parquet, generates 4-dimensional embeddings, and bulk-inserts them into the RDS PostgreSQL table.

  • Verify the Parquet output exists in S3 by running:
aws s3 ls s3://[[BUCKET_NAME="your-bucket-name"]]/processed/movies/
  • You should see one or more .parquet files listed in the output.
  • Download the processed Parquet files to a local directory by running:
mkdir -p processed_data
aws s3 cp s3://[[BUCKET_NAME="your-bucket-name"]]/processed/movies/ ./processed_data/ --recursive
  • Confirm the download by running:
ls processed_data/
  • You should see at least one .parquet file listed.

No files in the output?

Double check that your Glue job status is SUCCEEDED. If the job is still running, the processed prefix will be empty.

Also confirm your bucket name matches the one from terraform output s3_bucket_name.

Help me find my processed Parquet files in S3

Now you will write the Python script that reads these Parquet files, generates vector embeddings, and loads them into the RDS database.

  • Create the file scripts/load_vectors.py and add the imports and database configuration:
import os
import glob
import numpy as np
import psycopg2
from pgvector.psycopg2 import register_vector
import pyarrow.parquet as pq

DB_HOST = os.environ.get("DB_HOST", "your-rds-endpoint-here")
DB_NAME = os.environ.get("DB_NAME", "recommendations")
DB_USER = os.environ.get("DB_USER", "postgres")
DB_PASS = os.environ.get("DB_PASSWORD", "your-password-here")


def connect_db():
    conn = psycopg2.connect(
        host=DB_HOST.split(":")[0],
        port=5432,
        dbname=DB_NAME,
        user=DB_USER,
        password=DB_PASS
    )
    register_vector(conn)
    return conn


def setup_schema(conn):
    with conn.cursor() as cur:
        cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        cur.execute("""
            CREATE TABLE IF NOT EXISTS movies (
                movie_id INTEGER PRIMARY KEY,
                title TEXT NOT NULL,
                genre TEXT,
                avg_rating REAL,
                num_ratings INTEGER,
                year INTEGER,
                embedding vector(4)
            );
        """)
    conn.commit()

What does this code do?

The script reads database credentials from environment variables, falling back to placeholder defaults. The connect_db() function establishes a connection to your RDS instance and registers the pgvector type so Python can work with vector columns.

The setup_schema() function ensures the pgvector extension and movies table exist. This is idempotent, so it is safe to run multiple times.

  • Add the embedding generation and Parquet loading functions below the existing code:
def generate_embedding(norm_rating, norm_year, log_num_ratings, genre_index):
    return np.array(
        [norm_rating, norm_year, log_num_ratings, genre_index / 10.0],
        dtype=np.float32
    )


def load_from_parquet(conn, parquet_dir):
    parquet_files = glob.glob(os.path.join(parquet_dir, "*.parquet"))
    if not parquet_files:
        print("No parquet files found.")
        return

    table = pq.read_table(parquet_files[0])
    df = table.to_pandas()

    with conn.cursor() as cur:
        cur.execute("TRUNCATE TABLE movies;")
        for _, row in df.iterrows():
            embedding = generate_embedding(
                row["norm_rating"],
                row["norm_year"],
                row["log_num_ratings"],
                row["genre_index"]
            )
            cur.execute(
                """INSERT INTO movies
                   (movie_id, title, genre, avg_rating, num_ratings, year, embedding)
                   VALUES (%s, %s, %s, %s, %s, %s, %s)""",
                (
                    int(row["movie_id"]),
                    row["title"],
                    row["genre"],
                    float(row["avg_rating"]),
                    int(row["num_ratings"]),
                    int(row["year"]),
                    embedding.tolist()
                )
            )
    conn.commit()
    print(f"Loaded {len(df)} movies with embeddings.")

What does this code do?

The generate_embedding() function builds a 4-dimensional vector from the normalized features that the Glue job computed: rating, year, popularity (log-scaled), and genre index divided by 10.

The load_from_parquet() function reads the first Parquet file, converts it to a pandas DataFrame, truncates the existing table, and inserts each row with its computed embedding vector.

  • Add the similarity query function and the main entry point below the existing code:
def run_similarity_query(conn, movie_title):
    with conn.cursor() as cur:
        cur.execute(
            "SELECT embedding FROM movies WHERE title = %s",
            (movie_title,)
        )
        result = cur.fetchone()
        if not result:
            print(f"Movie '{movie_title}' not found.")
            return

        target_embedding = result[0]
        cur.execute(
            """SELECT title, genre, avg_rating, year,
                      embedding <=> %s AS distance
               FROM movies
               WHERE title != %s
               ORDER BY distance
               LIMIT 5""",
            (str(target_embedding), movie_title)
        )
        print(f"\nMovies similar to '{movie_title}':")
        print("-" * 50)
        for row in cur.fetchall():
            print(
                f"  {row[0]} ({row[3]}) - {row[1]} "
                f"- Rating: {row[2]:.1f} - Distance: {row[4]:.4f}"
            )


if __name__ == "__main__":
    conn = connect_db()
    setup_schema(conn)

    parquet_dir = "./processed_data"
    if os.path.exists(parquet_dir):
        load_from_parquet(conn, parquet_dir)
        run_similarity_query(conn, "Inception")
    else:
        print(f"Directory {parquet_dir} not found. Download from S3 first:")
        print("  aws s3 cp s3://<bucket>/processed/movies/ ./processed_data/ --recursive")

    conn.close()

What does this code do?

The run_similarity_query() function retrieves the embedding for a given movie title, then uses the pgvector cosine distance operator <=> to find the 5 most similar movies ranked by vector distance.

The __main__ block connects to the database, ensures the schema exists, loads vectors from the local Parquet directory, and immediately runs a similarity query for "Inception" to prove the pipeline works.

  • Save scripts/load_vectors.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before running it.

ⓧ I'd like to double check the full code

import os
import glob
import numpy as np
import psycopg2
from pgvector.psycopg2 import register_vector
import pyarrow.parquet as pq

DB_HOST = os.environ.get("DB_HOST", "your-rds-endpoint-here")
DB_NAME = os.environ.get("DB_NAME", "recommendations")
DB_USER = os.environ.get("DB_USER", "postgres")
DB_PASS = os.environ.get("DB_PASSWORD", "your-password-here")


def connect_db():
    conn = psycopg2.connect(
        host=DB_HOST.split(":")[0],
        port=5432,
        dbname=DB_NAME,
        user=DB_USER,
        password=DB_PASS
    )
    register_vector(conn)
    return conn


def setup_schema(conn):
    with conn.cursor() as cur:
        cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        cur.execute("""
            CREATE TABLE IF NOT EXISTS movies (
                movie_id INTEGER PRIMARY KEY,
                title TEXT NOT NULL,
                genre TEXT,
                avg_rating REAL,
                num_ratings INTEGER,
                year INTEGER,
                embedding vector(4)
            );
        """)
    conn.commit()


def generate_embedding(norm_rating, norm_year, log_num_ratings, genre_index):
    return np.array(
        [norm_rating, norm_year, log_num_ratings, genre_index / 10.0],
        dtype=np.float32
    )


def load_from_parquet(conn, parquet_dir):
    parquet_files = glob.glob(os.path.join(parquet_dir, "*.parquet"))
    if not parquet_files:
        print("No parquet files found.")
        return

    table = pq.read_table(parquet_files[0])
    df = table.to_pandas()

    with conn.cursor() as cur:
        cur.execute("TRUNCATE TABLE movies;")
        for _, row in df.iterrows():
            embedding = generate_embedding(
                row["norm_rating"],
                row["norm_year"],
                row["log_num_ratings"],
                row["genre_index"]
            )
            cur.execute(
                """INSERT INTO movies
                   (movie_id, title, genre, avg_rating, num_ratings, year, embedding)
                   VALUES (%s, %s, %s, %s, %s, %s, %s)""",
                (
                    int(row["movie_id"]),
                    row["title"],
                    row["genre"],
                    float(row["avg_rating"]),
                    int(row["num_ratings"]),
                    int(row["year"]),
                    embedding.tolist()
                )
            )
    conn.commit()
    print(f"Loaded {len(df)} movies with embeddings.")


def run_similarity_query(conn, movie_title):
    with conn.cursor() as cur:
        cur.execute(
            "SELECT embedding FROM movies WHERE title = %s",
            (movie_title,)
        )
        result = cur.fetchone()
        if not result:
            print(f"Movie '{movie_title}' not found.")
            return

        target_embedding = result[0]
        cur.execute(
            """SELECT title, genre, avg_rating, year,
                      embedding <=> %s AS distance
               FROM movies
               WHERE title != %s
               ORDER BY distance
               LIMIT 5""",
            (str(target_embedding), movie_title)
        )
        print(f"\nMovies similar to '{movie_title}':")
        print("-" * 50)
        for row in cur.fetchall():
            print(
                f"  {row[0]} ({row[3]}) - {row[1]} "
                f"- Rating: {row[2]:.1f} - Distance: {row[4]:.4f}"
            )


if __name__ == "__main__":
    conn = connect_db()
    setup_schema(conn)

    parquet_dir = "./processed_data"
    if os.path.exists(parquet_dir):
        load_from_parquet(conn, parquet_dir)
        run_similarity_query(conn, "Inception")
    else:
        print(f"Directory {parquet_dir} not found. Download from S3 first:")
        print("  aws s3 cp s3://<bucket>/processed/movies/ ./processed_data/ --recursive")

    conn.close()
Run the script and validate similarity results

The final step is to execute the loading script and confirm it successfully inserts vector embeddings and returns ranked similarity results.

  • Set the environment variables for your database connection. Replace the placeholders with your actual RDS endpoint and password:
export DB_HOST=[[RDS_ENDPOINT="your-rds-endpoint from terraform output"]]
export DB_PASSWORD=your-db-password-here

Where do I find my RDS endpoint?

Run terraform output rds_endpoint from inside your terraform/ directory. The value looks like recommendation-db.abc123.us-east-1.rds.amazonaws.com:5432. Use the full value including the port.

  • Run the vector loading script:
python3 scripts/load_vectors.py
  • You should see output confirming that 50 movies were loaded, followed by a list of 5 movies similar to "Inception" ranked by cosine distance:
Loaded 50 movies with embeddings.

Movies similar to 'Inception':
--------------------------------------------------
  The Matrix (1999) - SciFi - Rating: 4.7 - Distance: 0.0012
  ...

What do the results mean?

The <=> operator computes cosine distance between two vectors. A distance of 0 means identical vectors. The results show that movies with similar genre, rating, year, and popularity to "Inception" appear at the top of the list.

This proves your entire pipeline works end-to-end: raw CSV in S3 was transformed by Glue into feature-rich Parquet, loaded as vector embeddings into PostgreSQL, and queried using cosine similarity.

Connection refused or timeout?

Verify your security group allows inbound traffic on port 5432 from your current IP address. Your IP may have changed since you ran terraform apply. Update the my_ip_cidr variable in terraform.tfvars and re-apply if needed.

Also confirm your DB_HOST includes the full endpoint (with the :5432 port). The script splits on : to extract just the hostname.

Help me connect to my RDS instance from my local machine

Secret mission

Add a Glue Crawler and Query with Athena

Your pipeline writes processed Parquet to S3, but what if you could query that data directly with SQL without loading it into a database? In this secret mission, you'll add a Glue Crawler to auto-catalog your Parquet schema and run serverless SQL queries against it with Amazon Athena.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. The RDS instance is the main ongoing cost if you are outside the AWS Free Tier.

Cost warning

Your RDS PostgreSQL instance (db.t3.micro) costs approximately $0.017/hr (~$12/month) if your account is outside the Free Tier. S3 storage costs are negligible for 50 records. The Glue job only costs money when it runs ($0.44/DPU-hour).

If you are not planning to continue building on this project immediately, choose Pause or Delete below to avoid unexpected charges.

Resources you used:

  • S3 bucket with raw CSV, Glue script, and processed Parquet data.
  • RDS PostgreSQL instance (recommendation-db) with pgvector extension and movies table.
  • AWS Glue job (recommendation-etl-job).
  • IAM role (glue-etl-recommendation-role).
  • Security group (rds-recommendation-sg).
  • Glue Crawler and Catalog Database (if you completed the Secret Mission).
  • Local project directory with Terraform config, Glue script, sample data, and Python scripts.
  • Python virtual environment with installed packages.

Keep everything running

No action needed. Choose this if you are still actively building on this pipeline or plan to extend it soon.

  • Your RDS instance continues running and accruing charges (~$12/month outside Free Tier).
  • S3 storage costs are negligible for this data volume.
  • The Glue job definition and IAM role incur no charges while idle.
  • Your local project files and virtual environment remain on your machine.

Pause - I'll come back to this later

Stop the RDS instance to avoid compute charges while keeping your data intact. Storage charges still apply (~$2.30/month for 20 GB gp2).

  • Open the RDS console in AWS.
  • Select your recommendation-db instance.
  • Click Actions, then Stop temporarily.
  • Confirm the stop action.

RDS auto-restarts after 7 days

AWS automatically restarts stopped RDS instances after 7 consecutive days. If you plan to be away longer, set a reminder to stop it again or delete it entirely.

Delete - I don't want to use this again

Remove all AWS resources and local files. This is the recommended option after completing the project.

  • In your terminal, navigate to the terraform directory inside your project folder.
  • Destroy all AWS resources by running:
terraform destroy -auto-approve

This removes the S3 bucket (and all its contents thanks to force_destroy = true), the RDS instance, the Glue job, the IAM role, the security group, and the Glue Crawler and Catalog Database if you created them.

  • Wait for Terraform to confirm all resources are destroyed. The RDS instance takes a few minutes to delete.
  • Deactivate and remove your Python virtual environment:
deactivate
cd ~/Desktop
rm -rf aws-batch-etl-pipeline

Adjust the folder name if needed

Replace aws-batch-etl-pipeline with whatever you named your project folder in Step 1.

  • Verify the destroy succeeded by running:
aws s3 ls | grep recommendation

You should see no output, confirming your bucket is gone.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade batch ETL pipeline that transforms raw data into queryable vector embeddings, all managed as code with Terraform.

You've learned how to:

  • Provision a complete data pipeline infrastructure using Terraform (S3, RDS PostgreSQL, Glue, and IAM roles).
  • Write a PySpark ETL job running on AWS Glue 5.0 that cleans raw CSV data and outputs processed Parquet to a data lake.
  • Deploy a pgvector-enabled PostgreSQL database and run cosine similarity queries on vector embeddings to power a recommendation system.
  • Secret Mission: Added a Glue Crawler to auto-catalog Parquet data and queried it serverlessly with Athena.

Ready to quiz yourself?