How do I build a pipeline that loads CSV files from S3 into Redshift?

A common, reliable pattern is:

S3 (raw CSVs) → ETL/orchestration step → Redshift staging table → final tables

Here’s a practical way to build it.

---

1) Set up S3 for input files

• Put your CSVs in an S3 bucket/prefix, for example:
  • s3://my-bucket/data/incoming/
• Decide how files arrive:
  • manually
  • batch upload
  • app writes files
  • event-driven ingest

Tip: Use one file format and schema per folder/prefix to keep loading simple.

---

2) Create Redshift tables

Usually you load into a staging table first, then transform into the final table.

Example staging table:

CREATE TABLE stg_orders (
  order_id        VARCHAR(50),
  customer_id     VARCHAR(50),
  order_date      VARCHAR(50),
  amount          VARCHAR(50)
);

Then a final typed table:

CREATE TABLE orders (
  order_id      BIGINT,
  customer_id   BIGINT,
  order_date    DATE,
  amount        DECIMAL(12,2)
);

---

3) Give Redshift permission to read S3

Redshift needs an IAM role with access to the bucket.

Create/attach an IAM role

Grant permissions like:

• s3:GetObject
• s3:ListBucket

Attach the role to your Redshift cluster/workgroup.

Example policy (simplified):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:ListBucket"],
      "Resource": [
        "arn:aws:s3:::my-bucket",
        "arn:aws:s3:::my-bucket/data/incoming/*"
      ]
    }
  ]
}

---

4) Load CSVs with Redshift COPY

The main command is COPY.

COPY stg_orders
FROM 's3://my-bucket/data/incoming/'
IAM_ROLE 'arn:aws:iam::123456789012:role/MyRedshiftRole'
FORMAT AS CSV
IGNOREHEADER 1
DATEFORMAT 'auto'
TIMEFORMAT 'auto'
ACCEPTINVCHARS
BLANKSASNULL
EMPTYASNULL
TRUNCATECOLUMNS;

Useful options

• IGNOREHEADER 1 — skip the header row
• DATEFORMAT 'auto' / TIMEFORMAT 'auto' — parse dates/times
• BLANKSASNULL / EMPTYASNULL — handle missing values
• TRUNCATECOLUMNS — avoid failures on oversized strings
• ACCEPTINVCHARS — tolerate invalid UTF-8

---

5) Transform from staging to final tables

After loading raw strings, cast and clean data.

INSERT INTO orders
SELECT
  CAST(order_id AS BIGINT),
  CAST(customer_id AS BIGINT),
  CAST(order_date AS DATE),
  CAST(amount AS DECIMAL(12,2))
FROM stg_orders;

If you want idempotency, you may:

• DELETE existing rows for that batch/date
• use MERGE if your Redshift version supports it
• load into a new partition-like batch table and swap

---

6) Orchestrate the pipeline

You’ll want a scheduler to run:

1. file validation
2. COPY into staging
3. data quality checks
4. transform into final tables
5. cleanup/archive files

Common orchestration tools:

• AWS Glue
• Apache Airflow / MWAA
• AWS Step Functions
• Lambda + EventBridge for simpler use cases
• dbt for transformation logic after loading

---

7) Add data quality and monitoring

At minimum:

• check file exists and is non-empty
• validate row counts
• log failed rows from Redshift
• alert on COPY errors

Redshift provides load error tracking in system tables, e.g.:

• STL_LOAD_ERRORS
• STL_ERROR

Example:

SELECT *
FROM STL_LOAD_ERRORS
ORDER BY starttime DESC
LIMIT 20;

---

8) Handle bad records

You have a few options:

• reject the whole file
• load with permissive settings and review errors later
• separate “quarantine” files for bad data

For stricter control, load into staging as VARCHAR, then validate/cast in SQL.

---

9) Automate file arrival

If you want near-real-time processing:

• S3 upload triggers EventBridge
• EventBridge starts a Step Functions workflow or Lambda
• Lambda kicks off COPY or triggers Airflow/Glue job

For larger batch workflows, Airflow/Glue is usually better than direct Lambda loads.

---

10) Example end-to-end flow

1. CSV lands in s3://my-bucket/data/incoming/
2. Orchestrator starts pipeline
3. Redshift COPY loads into stg_orders
4. SQL cleans/casts into orders
5. Metrics/checks run
6. Processed file moved to s3://my-bucket/data/archive/

---

Minimal example script

If you want to run it from Python:

import psycopg2

conn = psycopg2.connect(
    dbname="dev",
    host="my-redshift-cluster.xxxxx.us-east-1.redshift.amazonaws.com",
    port=5439,
    user="awsuser",
    password="your_password"
)

cur = conn.cursor()

copy_sql = """
COPY stg_orders
FROM 's3://my-bucket/data/incoming/'
IAM_ROLE 'arn:aws:iam::123456789012:role/MyRedshiftRole'
FORMAT AS CSV
IGNOREHEADER 1
DATEFORMAT 'auto'
TIMEFORMAT 'auto';
"""

cur.execute(copy_sql)
conn.commit()
cur.close()
conn.close()

---

Best practices

• Use staging tables
• Keep schemas consistent
• Compress and split large files if possible
• Prefer Parquet over CSV if you can change the format
• Use IAM roles, not embedded AWS keys
• Monitor COPY errors and row counts
• Archive or delete processed files

---

If you want, I can also give you:

1. a production-ready architecture on AWS,
2. an Airflow DAG example, or
3. a Redshift COPY command tailored to your CSV format.