data-engineering-zoomcamp Documentation

Data Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 👇🏼

46,338 stars Jupyter Notebook 2 files · ~63,579 tokens #course#data-engineering#dbt#docker#free#kafka#kestra#spark
RAW Doc

Docker & Self-Hosted Deployment Verified OCI

Deploy data-engineering-zoomcamp on any server, homelab, or cloud container platform.

1-Click Terminal Command
bash (docker run) 
docker run -d --name data-engineering-zoomcamp -p 5432:5432 --restart always postgres:18
Configured Services & Ports
⚙️ pgdatabase
image: postgres:18
🔌 Ports: 5432:5432
⚙️ pgadmin
image: dpage/pgadmin4
🔌 Ports: 8085:80
⚙️ kestra_postgres
image: postgres:18
⚙️ kestra
image: kestra/kestra:v1.1
🔌 Ports: 8080:8080, 8081:8081
docker-compose.yml
yaml (docker-compose.yml) 
volumes:
  ny_taxi_postgres_data:
    driver: local
  kestra_postgres_data:
    driver: local
  kestra_data:
    driver: local
  kestra_tmp:
    driver: local

services:
  pgdatabase:
    image: postgres:18
    environment:
      POSTGRES_USER: root
      POSTGRES_PASSWORD: root
      POSTGRES_DB: ny_taxi
    ports:
      - "5432:5432"
    volumes:
      - ny_taxi_postgres_data:/var/lib/postgresql
    depends_on:
      kestra:
        condition: service_started

  pgadmin:
    image: dpage/pgadmin4
    environment:
      - [email protected]
      - PGADMIN_DEFAULT_PASSWORD=root
    ports:
      - "8085:80"
    depends_on:
      pgdatabase:
        condition: service_started

  kestra_postgres:
    image: postgres:18
    volumes:
      - kestra_postgres_data:/var/lib/postgresql
    environment:
      POSTGRES_DB: kestra
      POSTGRES_USER: kestra
      POSTGRES_PASSWORD: k3str4
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -d $${POSTGRES_DB} -U $${POSTGRES_USER}"]
      interval: 30s
      timeout: 10s
      retries: 10

  kestra:
    image: kestra/kestra:v1.1
    pull_policy: always
    # Note that this setup with a root user is intended for development purpose.
    # Our base image runs without root, but the Docker Compose implementation needs root to access the Docker socket
    # To run Kestra in a rootless mode in production, see: https://kestra.io/docs/installation/podman-compose
    user: "root"
    command: server standalone
    volumes:
      - kestra_data:/app/storage
      - /var/run/docker.sock:/var/run/docker.sock
      - kestra_tmp:/tmp/kestra-wd
    environment:
      KESTRA_CONFIGURATION: |
        datasources:
          postgres:
            url: jdbc:postgresql://kestra_postgres:5432/kestra
            driverClassName: org.postgresql.Driver
            username: kestra
            password: k3str4
        kestra:
          server:
            basicAuth:
              username: "[email protected]" # it must be a valid email address
              password: Admin1234!
          repository:
            type: postgres
          storage:
            type: local
            local:
              basePath: "/app/storage"
          queue:
            type: postgres
          tasks:
            tmpDir:
              path: /tmp/kestra-wd/tmp
          url: http://localhost:8080/
    ports:
      - "8080:8080"
      - "8081:8081"
    depends_on:
      kestra_postgres:
        condition: service_started

File: README.md

Data Engineering Zoomcamp: A Free 9-Week Course on Data Engineering Fundamentals

Resource Link
Course materials GitHub repository
Video lectures YouTube playlist
Documentation Zoomcamp Logistics · Data Engineering Zoomcamp
Course platform (deadlines, homework) courses.datatalks.club
Slack channel #course-data-engineering
Announcements Telegram
FAQ FAQ document

About the Course

This free 9-week course teaches the fundamentals of data engineering by building an end-to-end data pipeline from scratch. It consists of structured modules, hands-on workshops, and a final project, giving you practical experience with industry-standard tools and best practices.

Who Should Join

This course is for developers, analysts, and data scientists who want to learn how to build data pipelines and work with the modern data engineering stack. No prior data engineering experience is necessary.

Prerequisites

To get the most out of this course, you should have:

  • Basic coding experience
  • Familiarity with SQL
  • Experience with Python (helpful but not required)

No prior data engineering experience is necessary.

How to Take the Course

There are two ways to follow the course: live and self-paced.

Live Cohort Self-Paced
Start January 2027 Anytime
Lectures Pre-recorded Pre-recorded
Homework Graded Available but not scored
Leaderboard ✅ Yes ❌ No
Peer Review ✅ Yes ❌ No
Certificate ✅ Yes ❌ No
Cost Free Free
Register Sign up here Just start learning!

[!IMPORTANT]
"Live cohort" does not mean live classes. All lectures are pre-recorded. "Live" means working alongside others with deadlines, scored homework, a leaderboard, peer review, and a certificate at the end.

Self-paced steps:

  1. Follow the materials on GitHub
  2. Ask questions and share progress in Slack
  3. Do the homework (self-checked) and build a project for your portfolio

Syllabus

[Module 1: Containerization and Infrastructure as Code](01-docker-terraform/)

  • Introduction to GCP
  • Docker and Docker Compose
  • Running PostgreSQL with Docker
  • Infrastructure setup with Terraform
  • Homework

[Module 2: Workflow Orchestration](02-workflow-orchestration/)

  • Data Lakes and Workflow Orchestration
  • Workflow orchestration with Kestra
  • Homework

[Workshop 1: Data Ingestion](cohorts/2026/workshops/dlt.md)

  • API reading and pipeline scalability
  • Data normalization and incremental loading
  • Homework

[Module 3: Data Warehousing](03-data-warehouse/)

  • Introduction to BigQuery
  • Partitioning, clustering, and best practices
  • Machine learning in BigQuery

[Module 4: Analytics Engineering](04-analytics-engineering/)

  • Analytics Engineering and Data Modeling
  • dbt (data build tool) with DuckDB & BigQuery
  • Testing, documentation, and deployment

[Module 5: Data Platforms](05-data-platforms/)

  • Building end-to-end data pipelines with Bruin
  • Data ingestion, transformation, and quality
  • Deployment to cloud (BigQuery)

[Module 6: Batch Processing](06-batch/)

  • Introduction to Apache Spark
  • DataFrames and SQL
  • Internals of GroupBy and Joins

[Module 7: Streaming](07-streaming/)

  • Introduction to Kafka
  • Kafka Streams and KSQL
  • Schema management with Avro

Final Project

The final project applies all the concepts learned in a real-world scenario, including a peer review and feedback process.

Certificate

Certificates are awarded to learners who complete the final project during a live cohort. See Certification for how certification works and how to get your certificate.

Instructors

Past instructors:

Testimonials

Thank you for what you do! The Data Engineering Zoomcamp gave me skills that helped me land my first tech job.

— Tim Claytor (Source)

Three months might seem like a long time, but the growth and learning during this period are truly remarkable. It was a great experience with a lot of learning, connecting with like-minded people from all around the world, and having fun. I must admit, this was really hard. But the feeling of accomplishment and learning made it all worthwhile. And I would do it again!

— Nevenka Lukic (Source)

One of the significant things I inferred from the Zoomcamp is to prioritize fundamentals and principles over ever-evolving tools and tech stacks. Hugely grateful to Alexey Grigorev for putting together this incredible course and offering it for free.

— Siddhartha Gogoi (Source)

Such a fun deep dive into data engineering, cloud automation, and orchestration. I learned so much along the way. Big shoutout to Alexey Grigorev and the DataTalksClub team for the opportunity and guidance throughout the 3 months of the free course.

— Assitan NIARE (Source)

If you're serious about breaking into data engineering, start here. The repo's structure, community, and hands-on focus make it unparalleled.

— Wady Osama (Source)

Community & Support

Getting Help on Slack

Join the #course-data-engineering channel on DataTalks.Club Slack for discussions, troubleshooting, and networking.

To keep discussions organized:

Learning in Public

Share your progress as you go — see the learning in public guide.

Sponsors

A special thanks to our course sponsors for making this initiative possible!

Interested in supporting our community? Reach out to [email protected].

FAQ

A few common questions. For everything else, see the full Data Engineering Zoomcamp FAQ.

Q: Is this course really free?
A: Yes. All videos, materials, and homework are free and open-source.

Q: Do I need prior data engineering experience?
A: No. You just need basic coding experience and some familiarity with SQL. Python helps but isn't required.

Q: What does "live cohort" mean? Are there live classes?
A: No mandatory live classes. All lectures are pre-recorded. "Live" means deadlines, scored homework, a leaderboard, peer review, and certificate eligibility.

Q: Can I take it self-paced, and will I get a certificate?
A: Yes, you can start anytime. Certificates require completing the final project and peer reviews during a live cohort.

About DataTalks.Club

All the activity at DataTalks.Club mainly happens on Slack. We post updates there and discuss different aspects of data, career questions, and more.

At DataTalks.Club, we organize online events, community activities, and free courses. You can learn more about what we do at DataTalks.Club docs.

File: awesome-data-engineering.md

Have you found any cool resources about data engineering? Put them here

Learning Data Engineering

Courses

Books

Introduction to Data Engineering Terms

Data engineering in practice

Conference talks from companies, blog posts, etc

Doing Data Engineering

Coding & Python

SQL

Workflow orchestration

ETL and ELT

Data lakes

Data warehousing

Streaming

  • Building Streaming Analytics: The Journey and Learnings - Maxim Lukichev

DataOps

Monitoring and observability

Analytics engineering

Data mesh

Cloud

Reverse ETL

Career in Data Engineering

Data Engineering Management

Data engineering projects

Data Engineering Resources

Blogs

Podcasts

Communities

Meetups

People to follow on Twitter and LinkedIn

  • TODO

YouTube channels

Resource aggregators

License

This work is licensed under a Creative Commons Attribution 4.0 International License.

CC BY 4.0

File: certificates.md

Getting your certificate

Congratulations on finishing the course!

You can find your certificate in your enrollment profile (you need to be logged in):

If you can't find a certificate in your profile, it means you didn't pass the project.
If you believe it's a mistake, write in the course channel in Slack.

Adding to LinkedIn

You can add your certificate to LinkedIn:

  • Log in to your LinkedIn account, then go to your profile.
  • On the right, in the "Add profile" section dropdown, choose "Background" and then select the drop-down triangle next to "Licenses & Certifications".
  • In "Name", enter "Data Engineering Zoomcamp".
  • In "Issuing Organization", enter "DataTalksClub".
  • (Optional) In "Issue Date", enter the time when the certificate was created.
  • (Optional) Select the checkbox This certification does not expire.
  • Put your certificate ID.
  • In "Certification URL", enter the URL for your certificate.

Adapted from here

File: course.yaml

schema_version: 2
content_id: aeeff1c9-a209-4e0e-83bc-d9a98594aaa1
slug: de-zoomcamp
title: Data Engineering Zoomcamp
current_cohort: "2027"
cohorts:

  • identifier: "2027"
    content: root
  • identifier: "2026"
    content: cohorts/2026
    legacy: true
  • identifier: "2025"
    content: cohorts/2025
    legacy: true
  • identifier: "2024"
    content: cohorts/2024
    legacy: true
  • identifier: "2023"
    content: cohorts/2023
    legacy: true
  • identifier: "2022"
    content: cohorts/2022
    legacy: true
    Data Engineering Zoomcamp is a free, intensive course that takes you from data engineering fundamentals to building production-grade data pipelines with industry-standard tools. No previous data engineering experience is required.

The course runs for ten weeks: seven weeks of modules covering containerization and infrastructure as code, workflow orchestration, data warehousing, analytics engineering, batch processing, and streaming, followed by three weeks for a final project. Each module has video lectures, reading material, and a homework assignment.
outcome: Build production-grade data pipelines with industry-standard tools, starting from data engineering fundamentals.
starting_point: You are new to data engineering and want to understand how production data pipelines fit together.
prerequisites: You can write Python and SQL, use the command line, and have some Docker familiarity.
progression:

File: workshop-best-practices.md

Workshop Best Practices

Preferences and patterns learned from building the PyFlink streaming workshop.

Structure and Pacing

  • Introduce services one at a time, not all at once. Start with one container
    (e.g., Redpanda), explain it, use it. Then add the next (PostgreSQL), etc.
  • Start with the simplest version that works (plain Python consumer), then
    motivate the more complex tool (Flink) by showing what's missing.
  • Use docker compose up <service> -d to start services selectively during
    the gradual buildup. docker compose up --build -d only when everything
    is ready.

Data

  • Use real datasets, not fake test data. NYC taxi data
    (yellow_tripdata_YYYY-MM.parquet) is a good go-to.
  • Limit to manageable sizes (e.g., first 1000 rows) for workshop speed.

Project Setup

  • Assume starting from scratch: uv init -p 3.12 + uv add .
  • Add dependencies gradually as they're needed in the narrative
    (e.g., uv add kafka-python pandas pyarrow first, uv add psycopg2-binary
    later when PostgreSQL is introduced).
  • Always note "if you cloned the repo, run uv sync instead" as a blockquote.

Code Delivery

  • Break large code blocks into small, focused blocks. Each block should do
    one thing. Don't dump a full script in one block.
  • Pattern for code blocks: short intro line (what it does), then the code,
    then the explanation of how it works below. Don't put detailed
    explanations before the code - let the reader see the code first.
  • Keep imports local to each block - don't introduce all imports upfront.
    Each block should only import what it uses.
  • Introduce functions and utilities where they're first used, not earlier.
    For example, show dataclasses.asdict() in the block that calls it, not
    in the block that defines the dataclass.
  • When introducing a function, show a test with sample data before using it
    in the real code. For example, create a test binary string to verify a
    deserializer, then pass it to the consumer.
  • Prefer named functions over inline lambdas. A named function is reusable,
    testable, and easier to explain step by step. For example,
    value_deserializer=ride_deserializer instead of
    value_deserializer=lambda m: json.loads(m.decode('utf-8')).
  • Extract repetitive logic into named functions. For example, row-to-object
    conversion that appears in multiple places should be a function like
    ride_from_row(row).
  • Split one-liner functions into multiple lines. Each step (decode, parse,
    construct) on its own line is easier to follow and explain.
  • Show the simple approach first, then improve it. For example, show a
    generic json_serializer with manual dataclasses.asdict() calls, then
    introduce a specialized ride_serializer that handles the conversion
    internally. Let the student feel the friction before showing the fix.
  • Extract shared code (dataclasses, serializers, deserializers, converters)
    into shared modules (e.g., models.py) so multiple scripts can import
    from one place.
  • Reference the complete script at the end (e.g., "> The complete script is
    in src/producers/producer.py.").
  • For infrastructure files that are long or complex (Dockerfile, YAML configs),
    link to the file on GitHub and provide a short summary list of what it does.
    Use wget to download from the GitHub repo instead of asking students to
    type them.
  • Mention that students can run Python code in Jupyter notebooks
    (uv add jupyter, uv run jupyter lab) as an alternative to .py scripts.
    The small-block style maps naturally to notebook cells.
  • Flink jobs must remain as .py files (they're submitted to the cluster via
    docker compose exec). Add a note explaining this distinction.

Formatting

  • No bold formatting (**text**) in README files. Use plain text.
  • No em dashes. Use hyphens with spaces (-) instead.
  • Use python not python3.
  • Use docker compose not docker-compose.
  • Use uvx pgcli not just pgcli.
  • Use uv run python not python for running scripts.

Naming

  • Use meaningful names that reflect purpose, not generic placeholders.
    For example, group_id='rides-console' or group_id='rides-to-postgres',
    not group_id='test-consumer-group'.

Explanations

  • For complex configurations (like Redpanda's docker-compose command), explain
    every parameter in a table or list.
  • Explain the "why" not just the "what" (why two Kafka addresses? why
    checkpointing every 10 seconds? why watermarks?).
  • Use tables for parameter explanations and comparisons.
  • Include sample output for every command students will run.
  • Use > blockquotes for tips, notes about the repo, and common mistakes from
    original workshops/streams.
  • For complex concepts (watermarks, task slots, parallelism), pull the
    explanation out of bullet lists into its own multi-paragraph section. State
    the value or syntax in the bullet, then explain the concept below in
    separate paragraphs for easier reading.
  • Use lists for multi-point summaries instead of packing everything into one
    long sentence.
  • When showing a development shortcut (like mounting local files into Docker),
    add a note explaining how it works in production. Students benefit from
    understanding real-world deployment patterns alongside the workshop setup.

Code Organization

  • Define the source (where you read from) before the sink (where you write
    to) when presenting code blocks. Set up the consumer/reader first, then
    the database connection or output destination.

Docker Compose

  • Don't use container_name or hostname - Docker Compose handles naming
    automatically.
  • Don't use extra_hosts unless specifically needed.
  • Service names are automatically resolvable as hostnames within the Docker
    network.
  • Prefer short service names (e.g., redpanda not redpanda-1).
  • Keep restart: on-failure only for services that need it (like databases).

Dependencies and Versions

  • Always use the latest stable versions of images and libraries.
  • Pin exact versions for Flink and its connectors (they must match).
  • Use uv for everything Python-related (package management, running scripts,
    even installing Python itself inside Docker).
  • Prefer COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/ in Dockerfiles
    instead of apt-get install.

Workshop Header

  • Credit the original stream/video at the top with a link.
  • If the new video is not yet available, put "TBA" with a sign-up link
    (e.g., Luma).
  • Brief description of what we'll build and prerequisites.

Workshop Flow Template

  1. Introduce the first component (message broker, database, etc.)
  2. Set up with docker-compose (explain parameters)
  3. Create a simple producer/writer
  4. Create a simple consumer/reader
  5. Add a database, save data
  6. Show limitations of the simple approach
  7. Introduce the framework (Flink, Spark, etc.)
  8. Reproduce the simple case with the framework
  9. Do something the simple approach can't (aggregation, windowing)
  10. Explain advanced concepts (window types, offsets, etc.)
  11. Cleanup
  12. Q&A - questions and answers from the original stream. Include production
    deployment topics here rather than as standalone sections.

File: 01-docker-terraform/README.md

Containerization and Infrastructure as Code

Units

  1. Introduction to Docker
  2. Virtual Environments and Data Pipelines
  3. Dockerizing the Pipeline
  4. Running PostgreSQL with Docker
  5. NY Taxi Dataset and Data Ingestion
  6. Creating the Data Ingestion Script
  7. pgAdmin - Database Management Tool
  8. Dockerizing the Ingestion Script
  9. Docker Compose
  10. SQL Refresher
  11. Cleanup
  12. Terraform Overview
  13. GCP Overview

Homework

Setup guides

Workshop code

  • Data pipeline — the Dockerised ingestion pipeline built in units 3-9
  • Terraform — the Terraform configuration used in units 12-13

Community notes

Did you take notes? You can share them here

File: 01-docker-terraform/01-introduction.md

next_url: 02-virtual-environment.md

Introduction to Docker

Docker is a containerization software that allows us to isolate software in a similar way to virtual machines but in a much leaner way.

A Docker image is a snapshot of a container that we can define to run our software, or in this case our data pipelines. By exporting our Docker images to Cloud providers such as Amazon Web Services or Google Cloud Platform we can run our containers there.

Why Docker?

Docker provides the following advantages:

  • Reproducibility: Same environment everywhere
  • Isolation: Applications run independently
  • Portability: Run anywhere Docker is installed

They are used in many situations:

  • Integration tests: CI/CD pipelines
  • Running pipelines on the cloud: AWS Batch, Kubernetes jobs
  • Spark: Analytics engine for large-scale data processing
  • Serverless: AWS Lambda, Google Functions

Basic Docker Commands

Check Docker version:

bash
docker --version

Run a simple container:

bash
docker run hello-world

Run something more complex:

bash
docker run ubuntu

Nothing happens. Need to run it in -it mode:

bash
docker run -it ubuntu

We don't have python there so let's install it:

bash
apt update && apt install python3
python3 -V

Stateless Containers

Important: Docker containers are stateless - any changes done inside a container will NOT be saved when the container is killed and started again.

When you exit the container and use it again, the changes are gone:

bash
docker run -it ubuntu
python3 -V

This is good, because it doesn't affect your host system. Let's say you do something crazy like this:

bash
docker run -it ubuntu
rm -rf / # don't run it on your computer!

Next time we run it, all the files are back.

Managing Containers

But, this is not completely correct. The state is saved somewhere. We can see stopped containers:

bash
docker ps -a

We can restart one of them, but we won't do it, because it's not a good practice. They take space, so let's delete them:

bash
docker rm $(docker ps -aq)

Next time we run something, we add --rm:

bash
docker run -it --rm ubuntu

Different Base Images

There are other base images besides hello-world and ubuntu. For example, Python:

bash
docker run -it --rm python:3.9.16
# add -slim to get a smaller version

This one starts python. If we want bash, we need to overwrite entrypoint:

bash
docker run -it \
    --rm \
    --entrypoint=bash \
    python:3.9.16-slim

Volumes

So, we know that with docker we can restore any container to its initial state in a reproducible manner. But what about data? A common way to do so is with volumes.

The durable relationship is that host files can be shared with a container
through a mounted volume:

Let's create some data in test:

bash
mkdir test
cd test
touch file1.txt file2.txt file3.txt
echo "Hello from host" > file1.txt
cd ..

Now let's create a simple script test/list_files.py that shows the files in the folder:

python
from pathlib import Path

current_dir = Path.cwd()
current_file = Path(__file__).name

print(f"Files in {current_dir}:")

for filepath in current_dir.iterdir():
    if filepath.name == current_file:
        continue

    print(f"  - {filepath.name}")

    if filepath.is_file():
        content = filepath.read_text(encoding='utf-8')
        print(f"    Content: {content}")

Now let's map this to a Python container:

bash
docker run -it \
    --rm \
    -v $(pwd)/test:/app/test \
    --entrypoint=bash \
    python:3.9.16-slim

Inside the container, run:

bash
cd /app/test
ls -la
cat file1.txt
python list_files.py

You'll see the files from your host machine are accessible in the container!

File: 01-docker-terraform/02-virtual-environment.md

prev_url: 01-introduction.md next_url: 03-dockerizing-pipeline.md

Virtual Environments and Data Pipelines

A data pipeline is a service that receives data as input and outputs more data. For example, reading a CSV file, transforming the data somehow and storing it as a table in a PostgreSQL database.

mermaid
graph LR
    A[CSV File] --> B[Data Pipeline]
    B --> C[Parquet File]
    B --> D[PostgreSQL Database]
    B --> E[Data Warehouse]
    style B fill:#4CAF50,stroke:#333,stroke-width:2px,color:#fff

In this workshop, we'll build pipelines that:

  • Download CSV data from the web
  • Transform and clean the data with pandas
  • Load it into PostgreSQL for querying
  • Process data in chunks to handle large files

Creating a Simple Pipeline

Let's create an example pipeline. First, create a directory pipeline and inside, create a file pipeline.py:

python
import sys
print("arguments", sys.argv)

day = int(sys.argv[1])
print(f"Running pipeline for day {day}")

Now let's add pandas:

python
import pandas as pd

df = pd.DataFrame({"A": [1, 2], "B": [3, 4]})
print(df.head())

df.to_parquet(f"output_day_{sys.argv[1]}.parquet")

Why Virtual Environments?

We need pandas, but we don't have it. We want to test it before we run things in a container.

We can install it with pip:

bash
pip install pandas pyarrow

But this installs it globally on your system. This can cause conflicts if different projects need different versions of the same package.

Instead, we want to use a virtual environment - an isolated Python environment that keeps dependencies for this project separate from other projects and from your system Python.

Using uv - Modern Python Package Manager

We'll use uv - a modern, fast Python package and project manager written in Rust. It's much faster than pip and handles virtual environments automatically.

bash
pip install uv

Now initialize a Python project with uv:

bash
uv init --python=3.13

This creates a pyproject.toml file for managing dependencies and a .python-version file.

Comparing Python Versions

bash
uv run which python  # Python in the virtual environment
uv run python -V

which python        # System Python
python -V

You'll see they're different - uv run uses the isolated environment.

Adding Dependencies

Now let's add pandas:

bash
uv add pandas pyarrow

This adds pandas to your pyproject.toml and installs it in the virtual environment.

Running the Pipeline

Now we can execute the file:

bash
uv run python pipeline.py 10

We will see:

  • ['pipeline.py', '10']
  • job finished successfully for day = 10

Git Configuration

This script produces a binary (parquet) file, so let's make sure we don't accidentally commit it to git by adding parquet extensions to .gitignore:

text
*.parquet

File: 01-docker-terraform/03-dockerizing-pipeline.md

prev_url: 02-virtual-environment.md next_url: 04-postgres-docker.md

Dockerizing the Pipeline

Now let's containerize the script. Create the following Dockerfile file:

Simple Dockerfile with pip

dockerfile
# base Docker image that we will build on
FROM python:3.13.11-slim

# set up our image by installing prerequisites; pandas in this case
RUN pip install pandas pyarrow

# set up the working directory inside the container
WORKDIR /app
# copy the script to the container. 1st name is source file, 2nd is destination
COPY pipeline.py pipeline.py

# define what to do first when the container runs
# in this example, we will just run the script
ENTRYPOINT ["python", "pipeline.py"]

Explanation:

  • FROM: Base image (Python 3.13)
  • RUN: Execute commands during build
  • WORKDIR: Set working directory
  • COPY: Copy files into the image
  • ENTRYPOINT: Default command to run

Build and Run

Let's build the image:

bash
docker build -t test:pandas .
  • The image name will be test and its tag will be pandas. If the tag isn't specified it will default to latest.

We can now run the container and pass an argument to it, so that our pipeline will receive it:

bash
docker run -it test:pandas some_number

You should get the same output you did when you ran the pipeline script by itself.

Note: these instructions assume that pipeline.py and Dockerfile are in the same directory. The Docker commands should also be run from the same directory as these files.

Dockerfile with uv

What about uv? Let's use it instead of using pip:

dockerfile
# Start with slim Python 3.13 image
FROM python:3.13.10-slim

# Copy uv binary from official uv image (multi-stage build pattern)
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/

# Set working directory
WORKDIR /app

# Add virtual environment to PATH so we can use installed packages
ENV PATH="/app/.venv/bin:$PATH"

# Copy dependency files first (better layer caching)
COPY "pyproject.toml" "uv.lock" ".python-version" ./
# Install dependencies from lock file (ensures reproducible builds)
RUN uv sync --locked

# Copy application code
COPY pipeline.py pipeline.py

# Set entry point
ENTRYPOINT ["uv", "run", "python", "pipeline.py"]

File: 01-docker-terraform/04-postgres-docker.md

prev_url: 03-dockerizing-pipeline.md next_url: 05-data-ingestion.md

Running PostgreSQL with Docker

Now we want to do real data engineering. Let's use a Postgres database for that.

You can run a containerized version of Postgres that doesn't require any installation steps. You only need to provide a few environment variables to it as well as a volume for storing data.

Running PostgreSQL in a Container

Create a folder anywhere you'd like for Postgres to store data in. We will use the example folder ny_taxi_postgres_data. Here's how to run the container:

bash
docker run -it --rm \
  -e POSTGRES_USER="root" \
  -e POSTGRES_PASSWORD="root" \
  -e POSTGRES_DB="ny_taxi" \
  -v ny_taxi_postgres_data:/var/lib/postgresql \
  -p 5432:5432 \
  postgres:18

Explanation of Parameters

  • -e sets environment variables (user, password, database name)
  • -v ny_taxi_postgres_data:/var/lib/postgresql creates a named volume
    • Docker manages this volume automatically
    • Data persists even after container is removed
    • Volume is stored in Docker's internal storage
  • -p 5432:5432 maps port 5432 from container to host
  • postgres:18 uses PostgreSQL version 18 (latest as of Dec 2025)

Alternative Approach - Bind Mount

First create the directory, then map it:

bash
mkdir ny_taxi_postgres_data

docker run -it \
  -e POSTGRES_USER="root" \
  -e POSTGRES_PASSWORD="root" \
  -e POSTGRES_DB="ny_taxi" \
  -v $(pwd)/ny_taxi_postgres_data:/var/lib/postgresql \
  -p 5432:5432 \
  postgres:18

Named Volume vs Bind Mount

  • Named volume (name:/path): Managed by Docker, easier
  • Bind mount (/host/path:/container/path): Direct mapping to host filesystem, more control

Connecting to PostgreSQL

Once the container is running, we can log into our database with pgcli.

Install pgcli:

bash
uv add --dev pgcli

The --dev flag marks this as a development dependency (not needed in production). It will be added to the [dependency-groups] section of pyproject.toml instead of the main dependencies section.

Now use it to connect to Postgres:

bash
uv run pgcli -h localhost -p 5432 -u root -d ny_taxi
  • uv run executes a command in the context of the virtual environment
  • -h is the host. Since we're running locally we can use localhost.
  • -p is the port.
  • -u is the username.
  • -d is the database name.
  • The password is not provided; it will be requested after running the command.

When prompted, enter the password: root

Basic SQL Commands

Try some SQL commands:

sql
-- List tables
\dt

-- Create a test table
CREATE TABLE test (id INTEGER, name VARCHAR(50));

-- Insert data
INSERT INTO test VALUES (1, 'Hello Docker');

-- Query data
SELECT * FROM test;

-- Exit
\q

File: 01-docker-terraform/05-data-ingestion.md

prev_url: 04-postgres-docker.md next_url: 06-ingestion-script.md

NY Taxi Dataset and Data Ingestion

We will now create a Jupyter Notebook notebook.ipynb file which we will use to read a CSV file and export it to Postgres.

Setting up Jupyter

Install Jupyter:

bash
uv add --dev jupyter

Let's create a Jupyter notebook to explore the data:

bash
uv run jupyter notebook

The NYC Taxi Dataset

We will use data from the NYC TLC Trip Record Data website.

Specifically, we will use the Yellow taxi trip records CSV file for January 2021.

This data used to be csv, but later they switched to parquet. We want to keep using CSV because we need to do a bit of extra pre-processing (for the purposes of learning it).

A dictionary to understand each field is available here.

Note: The CSV data is stored as gzipped files. Pandas can read them directly.

Explore the Data

Create a new notebook and run:

python
import pandas as pd

# Read a sample of the data
prefix = 'https://github.com/DataTalksClub/nyc-tlc-data/releases/download/yellow/'
df = pd.read_csv(prefix + 'yellow_tripdata_2021-01.csv.gz', nrows=100)

# Display first rows
df.head()

# Check data types
df.dtypes

# Check data shape
df.shape

Note

  • When using nrows=100 to sample the first 100 rows, all columns have consistent data types in this subset, so the DtypeWarning below will NOT appear.
  • To reproduce the type warning shown below, remove the nrows=100 parameter and read the full dataset.

Handling Data Types

We have a warning: (Note that this warning might pop up later for some users, so it's best to follow the instructions below)

text
/tmp/ipykernel_25483/2933316018.py:1: DtypeWarning: Columns (6) have mixed types. Specify dtype option on import or set low_memory=False.

So we need to specify the types:

python
dtype = {
    "VendorID": "Int64",
    "passenger_count": "Int64",
    "trip_distance": "float64",
    "RatecodeID": "Int64",
    "store_and_fwd_flag": "string",
    "PULocationID": "Int64",
    "DOLocationID": "Int64",
    "payment_type": "Int64",
    "fare_amount": "float64",
    "extra": "float64",
    "mta_tax": "float64",
    "tip_amount": "float64",
    "tolls_amount": "float64",
    "improvement_surcharge": "float64",
    "total_amount": "float64",
    "congestion_surcharge": "float64"
}

parse_dates = [
    "tpep_pickup_datetime",
    "tpep_dropoff_datetime"
]

df = pd.read_csv(
    prefix + 'yellow_tripdata_2021-01.csv.gz',
    nrows=100,
    dtype=dtype,
    parse_dates=parse_dates
)

Ingesting Data into Postgres

In the Jupyter notebook, we create code to:

  1. Download the CSV file
  2. Read it in chunks with pandas
  3. Convert datetime columns
  4. Insert data into PostgreSQL using SQLAlchemy

Install SQLAlchemy

bash
uv add sqlalchemy "psycopg[binary,pool]"

Create Database Connection

python
from sqlalchemy import create_engine
engine = create_engine('postgresql+psycopg://root:root@localhost:5432/ny_taxi')

Get DDL Schema

python
print(pd.io.sql.get_schema(df, name='yellow_taxi_data', con=engine))

Output:

sql
CREATE TABLE yellow_taxi_data (
    "VendorID" BIGINT,
    tpep_pickup_datetime TIMESTAMP WITHOUT TIME ZONE,
    tpep_dropoff_datetime TIMESTAMP WITHOUT TIME ZONE,
    passenger_count BIGINT,
    trip_distance FLOAT(53),
    "RatecodeID" BIGINT,
    store_and_fwd_flag TEXT,
    "PULocationID" BIGINT,
    "DOLocationID" BIGINT,
    payment_type BIGINT,
    fare_amount FLOAT(53),
    extra FLOAT(53),
    mta_tax FLOAT(53),
    tip_amount FLOAT(53),
    tolls_amount FLOAT(53),
    improvement_surcharge FLOAT(53),
    total_amount FLOAT(53),
    congestion_surcharge FLOAT(53)
)

Create the Table

python
df.head(n=0).to_sql(name='yellow_taxi_data', con=engine, if_exists='replace')

head(n=0) makes sure we only create the table, we don't add any data yet.

Ingesting Data in Chunks

We don't want to insert all the data at once. Let's do it in batches and use an iterator for that:

python
df_iter = pd.read_csv(
    prefix + 'yellow_tripdata_2021-01.csv.gz',
    dtype=dtype,
    parse_dates=parse_dates,
    iterator=True,
    chunksize=100000
)

Iterate Over Chunks

python
for df_chunk in df_iter:
    print(len(df_chunk))

Inserting Data

python
df_chunk.to_sql(name='yellow_taxi_data', con=engine, if_exists='append')

Complete Ingestion Loop

python
first = True

for df_chunk in df_iter:

    if first:
        # Create table schema (no data)
        df_chunk.head(0).to_sql(
            name="yellow_taxi_data",
            con=engine,
            if_exists="replace"
        )
        first = False
        print("Table created")

    # Insert chunk
    df_chunk.to_sql(
        name="yellow_taxi_data",
        con=engine,
        if_exists="append"
    )

    print("Inserted:", len(df_chunk))

Alternative Approach (Without First Flag)

python
first_chunk = next(df_iter)

first_chunk.head(0).to_sql(
    name="yellow_taxi_data",
    con=engine,
    if_exists="replace"
)

print("Table created")

first_chunk.to_sql(
    name="yellow_taxi_data",
    con=engine,
    if_exists="append"
)

print("Inserted first chunk:", len(first_chunk))

for df_chunk in df_iter:
    df_chunk.to_sql(
        name="yellow_taxi_data",
        con=engine,
        if_exists="append"
    )
    print("Inserted chunk:", len(df_chunk))

Adding Progress Bar

Add tqdm to see progress:

bash
uv add tqdm

Put it around the iterable:

python
from tqdm.auto import tqdm

for df_chunk in tqdm(df_iter):
    ...

To see progress in terms of total chunks, you would have to add the total argument to tqdm(df_iter). In our scenario, the pragmatic way is
to hardcode a value based on the number of entries in the table.

Verify the Data

Connect to it using pgcli:

bash
uv run pgcli -h localhost -p 5432 -u root -d ny_taxi

And explore the data.

File: 01-docker-terraform/06-ingestion-script.md

prev_url: 05-data-ingestion.md next_url: 07-pgadmin.md

Creating the Data Ingestion Script

Now let's convert the notebook to a Python script.

Convert Notebook to Script

bash
uv run jupyter nbconvert --to=script notebook.ipynb
mv notebook.py ingest_data.py

The Complete Ingestion Script

See the pipeline/ directory for the complete script with click integration. Here's the core structure:

python
import pandas as pd
from sqlalchemy import create_engine
from tqdm.auto import tqdm

dtype = {
    "VendorID": "Int64",
    "passenger_count": "Int64",
    "trip_distance": "float64",
    "RatecodeID": "Int64",
    "store_and_fwd_flag": "string",
    "PULocationID": "Int64",
    "DOLocationID": "Int64",
    "payment_type": "Int64",
    "fare_amount": "float64",
    "extra": "float64",
    "mta_tax": "float64",
    "tip_amount": "float64",
    "tolls_amount": "float64",
    "improvement_surcharge": "float64",
    "total_amount": "float64",
    "congestion_surcharge": "float64"
}

parse_dates = [
    "tpep_pickup_datetime",
    "tpep_dropoff_datetime"
]

Click Integration

The script uses click for command-line argument parsing:

python
import click

@click.command()
@click.option('--pg-user', default='root', help='PostgreSQL user')
@click.option('--pg-pass', default='root', help='PostgreSQL password')
@click.option('--pg-host', default='localhost', help='PostgreSQL host')
@click.option('--pg-port', default=5432, type=int, help='PostgreSQL port')
@click.option('--pg-db', default='ny_taxi', help='PostgreSQL database name')
@click.option('--target-table', default='yellow_taxi_data', help='Target table name')
def run(pg_user, pg_pass, pg_host, pg_port, pg_db, target_table):
    # Ingestion logic here
    pass

Running the Script

The script reads data in chunks (100,000 rows at a time) to handle large files efficiently without running out of memory.

Example usage:

bash
uv run python ingest_data.py \
  --pg-user=root \
  --pg-pass=root \
  --pg-host=localhost \
  --pg-port=5432 \
  --pg-db=ny_taxi \
  --target-table=yellow_taxi_trips

File: 01-docker-terraform/07-pgadmin.md

prev_url: 06-ingestion-script.md next_url: 08-dockerizing-ingestion.md

pgAdmin - Database Management Tool

pgcli is a handy tool but it's cumbersome to use for complex queries and database management. pgAdmin is a web-based tool that makes it more convenient to access and manage our databases.

It's possible to run pgAdmin as a container along with the Postgres container, but both containers will have to be in the same virtual network so that they can find each other.

Run pgAdmin Container

bash
docker run -it \
  -e PGADMIN_DEFAULT_EMAIL="[email protected]" \
  -e PGADMIN_DEFAULT_PASSWORD="root" \
  -v pgadmin_data:/var/lib/pgadmin \
  -p 8085:80 \
  dpage/pgadmin4

The -v pgadmin_data:/var/lib/pgadmin volume mapping saves pgAdmin settings (server connections, preferences) so you don't have to reconfigure it every time you restart the container.

Parameters Explained

  • The container needs 2 environment variables: a login email and a password. We use [email protected] and root in this example.
  • pgAdmin is a web app and its default port is 80; we map it to 8085 in our localhost to avoid any possible conflicts.
  • The actual image name is dpage/pgadmin4.

Note: This won't work yet because pgAdmin can't see the PostgreSQL container. They need to be on the same Docker network!

Docker Networks

Let's create a virtual Docker network called pg-network:

bash
docker network create pg-network

You can remove the network later with the command docker network rm pg-network. You can look at the existing networks with docker network ls.

Run Containers on the Same Network

Stop both containers and re-run them with the network configuration:

bash
# Run PostgreSQL on the network
docker run -it \
  -e POSTGRES_USER="root" \
  -e POSTGRES_PASSWORD="root" \
  -e POSTGRES_DB="ny_taxi" \
  -v ny_taxi_postgres_data:/var/lib/postgresql \
  -p 5432:5432 \
  --network=pg-network \
  --name pgdatabase \
  postgres:18

# In another terminal, run pgAdmin on the same network
docker run -it \
  -e PGADMIN_DEFAULT_EMAIL="[email protected]" \
  -e PGADMIN_DEFAULT_PASSWORD="root" \
  -v pgadmin_data:/var/lib/pgadmin \
  -p 8085:80 \
  --network=pg-network \
  --name pgadmin \
  dpage/pgadmin4
  • Just like with the Postgres container, we specify a network and a name for pgAdmin.
  • The container names (pgdatabase and pgadmin) allow the containers to find each other within the network.

Connect pgAdmin to PostgreSQL

You should now be able to load pgAdmin on a web browser by browsing to http://localhost:8085. Use the same email and password you used for running the container to log in.

  1. Open browser and go to http://localhost:8085
  2. Login with email: [email protected], password: root
  3. Right-click "Servers" → Register → Server
  4. Configure:
    • General tab: Name: Local Docker
    • Connection tab:
      • Host: pgdatabase (the container name)
      • Port: 5432
      • Username: root
      • Password: root
  5. Save

Now you can explore the database using the pgAdmin interface!

File: 01-docker-terraform/08-dockerizing-ingestion.md

prev_url: 07-pgadmin.md next_url: 09-docker-compose.md

Dockerizing the Ingestion Script

Now let's containerize the ingestion script so we can run it in Docker.

The Dockerfile

The pipeline/Dockerfile shows how to containerize the ingestion script:

dockerfile
FROM python:3.13.11-slim
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/

WORKDIR /code
ENV PATH="/code/.venv/bin:$PATH"

COPY pyproject.toml .python-version uv.lock ./
RUN uv sync --locked

COPY ingest_data.py .

ENTRYPOINT ["uv", "run", "python", "ingest_data.py"]

Explanation

  • FROM python:3.13.11-slim: Start with slim Python 3.13 image for smaller size
  • COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/: Copy uv binary from official uv image
  • WORKDIR /code: Set working directory inside container
  • ENV PATH="/code/.venv/bin:$PATH": Add virtual environment to PATH
  • COPY pyproject.toml .python-version uv.lock ./: Copy dependency files first (better caching)
  • RUN uv sync --locked: Install all dependencies from lock file (ensures reproducible builds)
  • COPY ingest_data.py .: Copy ingestion script
  • ENTRYPOINT ["uv", "run", "python", "ingest_data.py"]: Set entry point to run the ingestion script

Build the Docker Image

bash
cd pipeline
docker build -t taxi_ingest:v001 .

Run the Containerized Ingestion

bash
docker run -it \
  --network=pg-network \
  taxi_ingest:v001 \
    --pg-user=root \
    --pg-pass=root \
    --pg-host=pgdatabase \
    --pg-port=5432 \
    --pg-db=ny_taxi \
    --target-table=yellow_taxi_trips

Important Notes

  • We need to provide the network for Docker to find the Postgres container. It goes before the name of the image.
  • Since Postgres is running on a separate container, the host argument will have to point to the container name of Postgres (pgdatabase).
  • You can drop the table in pgAdmin beforehand if you want, but the script will automatically replace the pre-existing table.

File: 01-docker-terraform/09-docker-compose.md

prev_url: 08-dockerizing-ingestion.md next_url: 10-sql-refresher.md

Docker Compose

docker-compose allows us to launch multiple containers using a single configuration file, so that we don't have to run multiple complex docker run commands separately.

Docker compose makes use of YAML files. Here's the docker-compose.yaml file:

yaml
services:
  pgdatabase:
    image: postgres:18
    environment:
      POSTGRES_USER: "root"
      POSTGRES_PASSWORD: "root"
      POSTGRES_DB: "ny_taxi"
    volumes:
      - "ny_taxi_postgres_data:/var/lib/postgresql"
    ports:
      - "5432:5432"

  pgadmin:
    image: dpage/pgadmin4
    environment:
      PGADMIN_DEFAULT_EMAIL: "[email protected]"
      PGADMIN_DEFAULT_PASSWORD: "root"
    volumes:
      - "pgadmin_data:/var/lib/pgadmin"
    ports:
      - "8085:80"

volumes:
  ny_taxi_postgres_data:
  pgadmin_data:

Explanation

  • We don't have to specify a network because docker compose takes care of it: every single container (or "service", as the file states) will run within the same network and will be able to find each other according to their names (pgdatabase and pgadmin in this example).
  • All other details from the docker run commands (environment variables, volumes and ports) are mentioned accordingly in the file following YAML syntax.

Start Services with Docker Compose

We can now run Docker compose by running the following command from the same directory where docker-compose.yaml is found. Make sure that all previous containers aren't running anymore:

bash
docker-compose up

Detached Mode

If you want to run the containers again in the background rather than in the foreground (thus freeing up your terminal), you can run them in detached mode:

bash
docker-compose up -d

Stop Services

You will have to press Ctrl+C in order to shut down the containers when running in foreground mode. The proper way of shutting them down is with this command:

bash
docker-compose down

Other Useful Commands

bash
# View logs
docker-compose logs

# Stop and remove volumes
docker-compose down -v

Benefits of Docker Compose

  • Single command to start all services
  • Automatic network creation
  • Easy configuration management
  • Declarative infrastructure

Running the Ingestion Script with Docker Compose

If you want to re-run the dockerized ingest script when you run Postgres and pgAdmin with docker compose, you will have to find the name of the virtual network that Docker compose created for the containers.

bash
# check the network link:
docker network ls

# it's pipeline_default (or similar based on directory name)
# now run the script:
docker run -it --rm\
  --network=pipeline_default \
  taxi_ingest:v001 \
    --pg-user=root \
    --pg-pass=root \
    --pg-host=pgdatabase \
    --pg-port=5432 \
    --pg-db=ny_taxi \
    --target-table=yellow_taxi_trips

File: 01-docker-terraform/10-sql-refresher.md

video_url: https://www.youtube.com/watch?v=QEcps_iskgg prev_url: 09-docker-compose.md next_url: 11-cleanup.md

SQL Refresher

Pre-Requisites: If you followed the course in the given order, Docker Compose should already be running with pgdatabase and pgAdmin.

Once done, you can go to http://localhost:8085/browser/ to access pgAdmin.
Don't forget to Right Click on the server or database to refresh it in case you don't see the new table.

Now start querying!

Inner Joins

Implicit INNER JOIN

Joining Yellow Taxi table with Zones Lookup table (implicit INNER JOIN):

sql
SELECT
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    total_amount,
    CONCAT(zpu."Borough", ' | ', zpu."Zone") AS "pickup_loc",
    CONCAT(zdo."Borough", ' | ', zdo."Zone") AS "dropoff_loc"
FROM
    yellow_taxi_trips t,
    zones zpu,
    zones zdo
WHERE
    t."PULocationID" = zpu."LocationID"
    AND t."DOLocationID" = zdo."LocationID"
LIMIT 100;

Explicit INNER JOIN

sql
SELECT
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    total_amount,
    CONCAT(zpu."Borough", ' | ', zpu."Zone") AS "pickup_loc",
    CONCAT(zdo."Borough", ' | ', zdo."Zone") AS "dropoff_loc"
FROM
    yellow_taxi_trips t
JOIN
-- or INNER JOIN but it's less used, when writing JOIN, postgreSQL understands implicitly that we want to use an INNER JOIN
    zones zpu ON t."PULocationID" = zpu."LocationID"
JOIN
    zones zdo ON t."DOLocationID" = zdo."LocationID"
LIMIT 100;

Data Quality Checks

Checking for NULL Location IDs

sql
SELECT
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    total_amount,
    "PULocationID",
    "DOLocationID"
FROM
    yellow_taxi_trips
WHERE
    "PULocationID" IS NULL
    OR "DOLocationID" IS NULL
LIMIT 100;

Checking for Location IDs NOT IN Zones Table

sql
SELECT
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    total_amount,
    "PULocationID",
    "DOLocationID"
FROM
    yellow_taxi_trips
WHERE
    "DOLocationID" NOT IN (SELECT "LocationID" from zones)
    OR "PULocationID" NOT IN (SELECT "LocationID" from zones)
LIMIT 100;

LEFT, RIGHT, and OUTER JOINS

Using LEFT, RIGHT, and OUTER JOINS when some Location IDs are not in either Tables:

sql
DELETE FROM zones WHERE "LocationID" = 142;

SELECT
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    total_amount,
    CONCAT(zpu."Borough", ' | ', zpu."Zone") AS "pickup_loc",
    CONCAT(zdo."Borough", ' | ', zdo."Zone") AS "dropoff_loc"
FROM
    yellow_taxi_trips t
LEFT JOIN
    zones zpu ON t."PULocationID" = zpu."LocationID"
JOIN
    zones zdo ON t."DOLocationID" = zdo."LocationID"
LIMIT 100;
sql
SELECT
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    total_amount,
    CONCAT(zpu."Borough", ' | ', zpu."Zone") AS "pickup_loc",
    CONCAT(zdo."Borough", ' | ', zdo."Zone") AS "dropoff_loc"
FROM
    yellow_taxi_trips t
RIGHT JOIN
    zones zpu ON t."PULocationID" = zpu."LocationID"
JOIN
    zones zdo ON t."DOLocationID" = zdo."LocationID"
LIMIT 100;
sql
SELECT
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    total_amount,
    CONCAT(zpu."Borough", ' | ', zpu."Zone") AS "pickup_loc",
    CONCAT(zdo."Borough", ' | ', zdo."Zone") AS "dropoff_loc"
FROM
    yellow_taxi_trips t
OUTER JOIN
    zones zpu ON t."PULocationID" = zpu."LocationID"
JOIN
    zones zdo ON t."DOLocationID" = zdo."LocationID"
LIMIT 100;

GROUP BY

Calculate Number of Trips Per Day

sql
SELECT
    CAST(tpep_dropoff_datetime AS DATE) AS "day",
    COUNT(1)
FROM
    yellow_taxi_trips
GROUP BY
    CAST(tpep_dropoff_datetime AS DATE)
LIMIT 100;

ORDER BY

Ordering by Day

sql
SELECT
    CAST(tpep_dropoff_datetime AS DATE) AS "day",
    COUNT(1)
FROM
    yellow_taxi_trips
GROUP BY
    CAST(tpep_dropoff_datetime AS DATE)
ORDER BY
    "day" ASC
LIMIT 100;

Ordering by Count

sql
SELECT
    CAST(tpep_dropoff_datetime AS DATE) AS "day",
    COUNT(1) AS "count"
FROM
    yellow_taxi_trips
GROUP BY
    CAST(tpep_dropoff_datetime AS DATE)
ORDER BY
    "count" DESC
LIMIT 100;

Other Aggregations

sql
SELECT
    CAST(tpep_dropoff_datetime AS DATE) AS "day",
    COUNT(1) AS "count",
    MAX(total_amount) AS "total_amount",
    MAX(passenger_count) AS "passenger_count"
FROM
    yellow_taxi_trips
GROUP BY
    CAST(tpep_dropoff_datetime AS DATE)
ORDER BY
    "count" DESC
LIMIT 100;

Grouping by Multiple Fields

sql
SELECT
    CAST(tpep_dropoff_datetime AS DATE) AS "day",
    "DOLocationID",
    COUNT(1) AS "count",
    MAX(total_amount) AS "total_amount",
    MAX(passenger_count) AS "passenger_count"
FROM
    yellow_taxi_trips
GROUP BY
    1, 2
ORDER BY
    "day" ASC,
    "DOLocationID" ASC
LIMIT 100;

File: 01-docker-terraform/11-cleanup.md

prev_url: 10-sql-refresher.md next_url: 12-terraform-overview.md

Cleanup

When you're done with the workshop, clean up Docker resources to free up disk space.

Stop All Running Containers

bash
docker-compose down

Remove Specific Containers

bash
# List all containers
docker ps -a

# Remove specific container
docker rm <container_id>

# Remove all stopped containers
docker container prune

Remove Docker Images

bash
# List all images
docker images

# Remove specific image
docker rmi taxi_ingest:v001

# Remove all unused images
docker image prune -a

Remove Docker Volumes

bash
# List volumes
docker volume ls

# Remove specific volumes
docker volume rm ny_taxi_postgres_data
docker volume rm pgadmin_data

# Remove all unused volumes
docker volume prune

Remove Docker Networks

bash
# List networks
docker network ls

# Remove specific network
docker network rm pg-network

# Remove all unused networks
docker network prune

Complete Cleanup

Removes ALL Docker resources - use with caution!

bash
# ⚠️ Warning: This removes ALL Docker resources!
docker system prune -a --volumes

Clean Up Local Files

bash
# Remove parquet files
rm *.parquet

# Remove Python cache
rm -rf __pycache__ .pytest_cache

# Remove virtual environment (if using venv)
rm -rf .venv

That's all for today. Happy learning! 🐳📊

File: 01-docker-terraform/12-terraform-overview.md

video_url: https://www.youtube.com/watch?v=18jIzE41fJ4 prev_url: 11-cleanup.md next_url: 13-gcp-overview.md

Terraform Overview

Concepts

Introduction

  1. What is Terraform?
    • open-source tool by HashiCorp, used for provisioning infrastructure resources
    • supports DevOps best practices for change management
    • Managing configuration files in source control to maintain an ideal provisioning state
      for testing and production environments
  2. What is IaC?
    • Infrastructure-as-Code
    • build, change, and manage your infrastructure in a safe, consistent, and repeatable way
      by defining resource configurations that you can version, reuse, and share.
  3. Some advantages
    • Infrastructure lifecycle management
    • Version control commits
    • Very useful for stack-based deployments, and with cloud providers such as AWS, GCP, Azure, K8S…
    • State-based approach to track resource changes throughout deployments

Terraform runs locally and uses a provider to plan and apply changes in a
cloud platform:

Files

  • main.tf
  • variables.tf
  • Optional: resources.tf, output.tf
  • .tfstate

Declarations

  • terraform: configure basic Terraform settings to provision your infrastructure
    • required_version: minimum Terraform version to apply to your configuration
    • backend: stores Terraform's "state" snapshots, to map real-world resources to your configuration.
      • local: stores state file locally as terraform.tfstate
    • required_providers: specifies the providers required by the current module
  • provider:
    • adds a set of resource types and/or data sources that Terraform can manage
    • The Terraform Registry is the main directory of publicly available providers from most major infrastructure platforms.
  • resource
    • blocks to define components of your infrastructure
    • Project modules/resources: google_storage_bucket, google_bigquery_dataset, google_bigquery_table
  • variable & locals
    • runtime arguments and constants

Execution steps

  1. terraform init:
    • Initializes & configures the backend, installs plugins/providers, & checks out an existing configuration from a version control
  2. terraform plan:
    • Matches/previews local changes against a remote state, and proposes an Execution Plan.
  3. terraform apply:
    • Asks for approval to the proposed plan, and applies changes to cloud
  4. terraform destroy
    • Removes your stack from the Cloud

Terraform Workshop to create GCP Infra

Continue here: week_1_basics_n_setup/1_terraform_gcp/terraform

References

https://learn.hashicorp.com/collections/terraform/gcp-get-started

File: 01-docker-terraform/13-gcp-overview.md

video_url: https://www.youtube.com/watch?v=18jIzE41fJ4 prev_url: 12-terraform-overview.md next_url: ../02-workflow-orchestration/01-what-is-workflow-orchestration.md

GCP Overview

Project infrastructure modules in GCP:

  • Google Cloud Storage (GCS): Data Lake
  • BigQuery: Data Warehouse

Cloud platforms group their services into families such as compute,
networking, storage, big data, identity, and machine learning:

(Concepts explained in Week 2 - Data Ingestion)

Initial Setup

For this course, we'll use a free version (upto EUR 300 credits).

  1. Create an account with your Google email ID
  2. Setup your first project if you haven't already
    • eg. "DTC DE Course", and note down the "Project ID" (we'll use this later when deploying infra with TF)
  3. Setup service account & authentication for this project
    • Grant Viewer role to begin with.
    • Download service-account-keys (.json) for auth.
  4. Download SDK for local setup
  5. Set environment variable to point to your downloaded GCP keys:
    shell
    export GOOGLE_APPLICATION_CREDENTIALS="<path/to/your/service-account-authkeys>.json"

# Refresh token/session, and verify authentication
gcloud auth application-default login

Setup for Access

  1. IAM Roles for Service account:

    • Go to the IAM section of IAM & Admin https://console.cloud.google.com/iam-admin/iam
    • Click the Edit principal icon for your service account.
    • Add these roles in addition to Viewer : Storage Admin + Storage Object Admin + BigQuery Admin

  2. Enable these APIs for your project:

  3. Please ensure GOOGLE_APPLICATION_CREDENTIALS env-var is set.

    shell
    export GOOGLE_APPLICATION_CREDENTIALS="<path/to/your/service-account-authkeys>.json"

Terraform Workshop to create GCP Infra

Continue here: week_1_basics_n_setup/1_terraform_gcp/terraform

File: 01-docker-terraform/module.yaml

schema_version: 2
content_id: c7a2f290-c7f0-4393-9914-f26280a042c6
title: "Containerization and Infrastructure as Code"
summary: "Run data pipelines and PostgreSQL in Docker, then provision cloud infrastructure with Terraform."
units:

  • content_id: 940c1ee5-8a9c-4485-b944-09afbc030e5b
    title: "Introduction to Docker"
    path: 01-introduction.md
  • content_id: bd43e3cd-e623-463b-b9d2-caf174f10297
    title: "Virtual Environments and Data Pipelines"
    path: 02-virtual-environment.md
  • content_id: d8cf9b9e-9db6-4960-8e13-bc191d13f26a
    title: "Dockerizing the Pipeline"
    path: 03-dockerizing-pipeline.md
  • content_id: f81b8dd5-5a8f-4f17-9d60-888d121e46db
    title: "Running PostgreSQL with Docker"
    path: 04-postgres-docker.md
  • content_id: ed350e99-a71d-4040-95c7-f5786f2e0dbe
    title: "NY Taxi Dataset and Data Ingestion"
    path: 05-data-ingestion.md
  • content_id: 3e60638a-0289-4b70-8af3-d8fe943935ea
    title: "Creating the Data Ingestion Script"
    path: 06-ingestion-script.md
  • content_id: 0ea9f94f-4482-45fa-a27b-e6339a7b4fec
    title: "pgAdmin - Database Management Tool"
    path: 07-pgadmin.md
  • content_id: d61fdb67-8a73-4c9a-ad88-3d2adcdb324f
    title: "Dockerizing the Ingestion Script"
    path: 08-dockerizing-ingestion.md
  • content_id: 49b47814-16dc-4578-ab7b-a06adb151a84
    title: "Docker Compose"
    path: 09-docker-compose.md
  • content_id: 684b5291-406c-4bb3-9096-94d888be63d4
    title: "SQL Refresher"
    path: 10-sql-refresher.md
  • content_id: ed009f0a-0de9-463f-af3e-c90bf94ce7e4
    title: "Cleanup"
    path: 11-cleanup.md
  • content_id: 5c87355c-17e6-44c0-b148-7f0113e38881
    title: "Terraform Overview"
    path: 12-terraform-overview.md
  • content_id: 72d444af-5b90-4d40-9204-630d0e11aa1a
    title: "GCP Overview"
    path: 13-gcp-overview.md

File: 01-docker-terraform/images/cloud-service-families.json

{
"title": "Cloud service families",
"description": "Cloud services grouped into compute, management, networking, storage and databases, big data, identity and security, and machine learning families.",
"canvas": {
"width": 600,
"height": 660,
"background": "#ffffff"
},
"layout": {
"type": "grid",
"card_width": 460,
"card_height": 64,
"row_gap": 18
},
"nodes": [
{
"id": "compute",
"title": "Compute",
"subtitle": "Compute Engine, GKE, App Engine",
"icon": "container",
"row": 0,
"column": 0,
"color": "purple"
},
{
"id": "management",
"title": "Management",
"subtitle": "Monitoring and admin tools",
"icon": "settings",
"color": "purple",
"row": 1,
"column": 0
},
{
"id": "networking",
"title": "Networking",
"subtitle": "VPC, load balancers, DNS",
"icon": "websocket",
"color": "purple",
"row": 2,
"column": 0
},
{
"id": "storage",
"title": "Storage & Databases",
"subtitle": "GCS, BigQuery, Cloud SQL",
"icon": "database",
"color": "purple",
"row": 3,
"column": 0
},
{
"id": "bigdata",
"title": "Big Data",
"subtitle": "Dataflow, Pub/Sub, Dataproc",
"icon": "workflow",
"color": "purple",
"row": 4,
"column": 0
},
{
"id": "identity",
"title": "Identity & Security",
"subtitle": "IAM and access management",
"icon": "shield",
"color": "purple",
"row": 5,
"column": 0
},
{
"id": "ml",
"title": "Machine Learning",
"subtitle": "Vertex AI and pre-trained APIs",
"icon": "sparkles",
"color": "purple",
"row": 6,
"column": 0
}
],
"edges": []
}

File: 01-docker-terraform/images/docker-volume-mapping.json

{
"title": "Docker volume mapping",
"description": "Local host files can be shared with a Docker container through a mounted volume.",
"canvas": {"width": 940, "height": 280, "background": "#ffffff"},
"layout": {"type": "manual", "card_width": 240, "card_height": 100},
"nodes": [
{"id": "host", "title": "Local host files", "subtitle": "code and data on your machine", "color": "blue", "icon": "document", "x": 50, "y": 90},
{"id": "volume", "title": "Mounted volume", "subtitle": "shared folder", "color": "purple", "icon": "volume", "x": 350, "y": 90},
{"id": "container", "title": "Docker container", "subtitle": "sees the same files", "color": "green", "icon": "container", "x": 650, "y": 90}
],
"edges": [
{"from": "host", "to": "volume", "bidirectional": true},
{"from": "volume", "to": "container", "bidirectional": true}
]
}

File: 01-docker-terraform/images/terraform-provider-flow.json

{
"title": "Terraform provider flow",
"description": "Terraform runs on a local machine and manages resources in a cloud provider.",
"canvas": {
"width": 1240,
"height": 280,
"background": "#ffffff"
},
"layout": {
"type": "manual",
"card_width": 240,
"card_height": 100
},
"nodes": [
{
"id": "local",
"title": "Local machine",
"subtitle": "where you run Terraform",
"color": "blue",
"icon": "user",
"x": 40,
"y": 90
},
{
"id": "tf",
"title": "Terraform",
"subtitle": "infrastructure as code",
"color": "amber",
"icon": "settings",
"x": 480,
"y": 90
},
{
"id": "cloud",
"title": "Cloud provider",
"subtitle": "GCP \u00b7 AWS \u00b7 Azure",
"color": "green",
"icon": "environment",
"x": 920,
"y": 90
}
],
"edges": [
{
"from": "local",
"to": "tf"
},
{
"from": "tf",
"to": "cloud",
"label": "plan & apply",
"from_anchor": "right_top",
"to_anchor": "left_top"
},
{
"from": "cloud",
"to": "tf",
"label": "read state",
"dashed": true,
"from_anchor": "left_bottom",
"to_anchor": "right_bottom"
}
]
}

File: 01-docker-terraform/pipeline/docker-compose.yaml

services:
pgdatabase:
image: postgres:18
environment:
POSTGRES_USER: "root"
POSTGRES_PASSWORD: "root"
POSTGRES_DB: "ny_taxi"
volumes:
- ny_taxi_postgres_data:/var/lib/postgresql
ports:
- "5432:5432"

pgadmin:
image: dpage/pgadmin4
environment:
PGADMIN_DEFAULT_EMAIL: "[email protected]"
PGADMIN_DEFAULT_PASSWORD: "root"
volumes:
- pgadmin_data:/var/lib/pgadmin
ports:
- "8085:80"

volumes:
ny_taxi_postgres_data:
pgadmin_data:

File: 01-docker-terraform/setup/terraform-and-gcp-setup.md

Local Setup for Terraform and GCP

Pre-Requisites

  1. Terraform client installation: https://www.terraform.io/downloads
  2. Cloud Provider account: https://console.cloud.google.com/

Terraform Concepts

Terraform Overview

GCP setup

  1. Setup for First-time
  2. IAM / Access specific to this course

Terraform Workshop for GCP Infra

Your setup is ready!
Now head to the terraform directory, and perform the execution steps to create your infrastructure.

File: 01-docker-terraform/setup/terraform-on-windows.md

GCP and Terraform on Windows

You don't need these instructions if you use WSL. It's only for "plain Windows"

Google Cloud SDK

When installing it, you might see something like that:

text
The installer is unable to automatically update your system PATH. Please add
  C:\tools\google-cloud-sdk\bin

Now we need to point it to correct Python installation. Assuming you use Anaconda:

bash
export CLOUDSDK_PYTHON=~/Anaconda3/python

Now let's check that it works:

bash
$ gcloud version
Google Cloud SDK 367.0.0
bq 2.0.72
core 2021.12.10
gsutil 5.5

Google Cloud SDK Authentication

  • Now create a service account and generate keys like shown in the videos
  • Download the key and put it to some location, e.g. .gc/ny-rides.json
  • Set GOOGLE_APPLICATION_CREDENTIALS to point to the file
bash
export GOOGLE_APPLICATION_CREDENTIALS=~/.gc/ny-rides.json

Now authenticate:

bash
gcloud auth activate-service-account --key-file $GOOGLE_APPLICATION_CREDENTIALS

Alternatively, you can authenticate using OAuth like shown in the video

bash
gcloud auth application-default login

If you get a message like quota exceeded

WARNING:
Cannot find a quota project to add to ADC. You might receive a "quota exceeded" or "API not enabled" error.
Run $ gcloud auth application-default set-quota-project to add a quota project.

Then run this:

bash
PROJECT_NAME="ny-rides-alexey"
gcloud auth application-default set-quota-project ${PROJECT_NAME}

Terraform

bash
terraform init

Optionally you can configure your terraform files (variables.tf) to include your project id:

bash
variable "project" {
  description = "Your GCP Project ID"
  default = "ny-rides-alexey"
  type = string
}

If you get an error like that:

Error: googleapi: Error 403: [email protected] does not have
storage.buckets.create access to the Google Cloud project., forbidden

Then you need to give your service account all the permissions. Make sure you follow the instructions in the videos

  • You can also use this file, but it doesn't list all the required permissions

File: 01-docker-terraform/terraform/README.md

Concepts

  • Terraform_overview
  • If you were unable to generate a service account keyfile due to organizational policies, refer to the instructions below

Execution

shell
# Refresh service-account's auth-token for this session
gcloud auth application-default login

# Initialize state file (.tfstate)
terraform init

# Check changes to new infra plan
terraform plan -var="project=<your-gcp-project-id>"
shell
# Create new infra
terraform apply -var="project=<your-gcp-project-id>"
shell
# Delete infra after your work, to avoid costs on any running services
terraform destroy

Warning

Remember to use a proper gitignore file before publishing your code on GitHub

Fallback

  1. Give yourself the token creator role on the pertinent service account

    bash
    gcloud iam service-accounts add-iam-policy-binding \
    <SERVICE_ACCOUNT_EMAIL> \
    --member="user:[email protected]" \
    --role="roles/iam.serviceAccountTokenCreator"

  2. Add the sections below the first block to your main terraform configuration

    terraform
    # Connect to gcp using ADC (identity verification)
 provider "google" {
   project = var.project
   region  = var.region
   zone    = var.zone
 }

 /* add these data blocks */
 
 # This data source gets a temporary token for the service account
 data "google_service_account_access_token" "default" {
   provider               = google
   target_service_account = "<SERVICE_ACCOUNT_EMAIL>"
   scopes                 = ["https://www.googleapis.com/auth/cloud-platform"]
   lifetime               = "3600s"
 }
 
 # This second provider block uses that temporary token and does the real work
 provider "google" {
   alias        = "impersonated"
   access_token = data.google_service_account_access_token.default.access_token
   project      = var.project
   region       = var.region
   zone         = var.zone
 }

  3. Now, you can follow the instructions above

File: 01-docker-terraform/terraform/terraform_with_variable_AWS/README.md

AWS Terraform Data Lake (GCP Equivalent)

📌 Overview

This repository contains an AWS-based Terraform implementation that mirrors the Google Cloud Platform (GCP) infrastructure used in the Data Engineering course (e.g. GCS + BigQuery), but implemented using AWS services.

The goal is to help learners who:

  • Are enrolled in a GCP-focused Data Engineering course
  • Prefer or need to work with AWS
  • Want to understand cloud-agnostic data engineering concepts

This setup focuses on building a basic data lake foundation using:

  • Amazon S3 (equivalent to GCS)
  • AWS Glue Data Catalog (equivalent to BigQuery datasets / metadata layer)
  • Terraform as Infrastructure as Code (IaC)

🏗️ Architecture Mapping (GCP → AWS)

GCP Service AWS Equivalent Purpose
Google Cloud Storage (GCS) Amazon S3 Data Lake storage
Uniform Bucket Level Access S3 Public Access Block Secure bucket access
Object Lifecycle Rules S3 Lifecycle Configuration Automatic data expiration
BigQuery Dataset AWS Glue Catalog Database Metadata & query layer
Terraform (GCP provider) Terraform (AWS provider) Infrastructure as Code

📁 Project Structure

text
.
├── main.tf            # Core infrastructure resources
├── variables.tf       # Input variable definitions
├── terraform.tfvars   # Environment-specific values
└── README.md          # Project documentation

---

## File: 02-workflow-orchestration/README.md

# Workflow Orchestration

## Units

1. [What is Workflow Orchestration?](01-what-is-workflow-orchestration.md)
2. [What is Kestra?](02-what-is-kestra.md)
3. [Installing Kestra](03-installing-kestra.md)
4. [Kestra Concepts](04-kestra-concepts.md)
5. [Orchestrate Python Code](05-orchestrate-python-code.md)
6. [Getting Started Pipeline](06-getting-started-pipeline.md)
7. [Local DB: Load Taxi Data to Postgres](07-load-taxi-data-to-postgres.md)
8. [Local DB: Learn Scheduling and Backfills](08-scheduling-and-backfills.md)
9. [ETL vs ELT](09-etl-vs-elt.md)
10. [Setup Google Cloud Platform (GCP)](10-setup-google-cloud-platform.md)
11. [GCP Workflow: Load Taxi Data to BigQuery](11-load-taxi-data-to-bigquery.md)
12. [GCP Workflow: Schedule and Backfill Full Dataset](12-schedule-and-backfill-full-dataset.md)
13. [Introduction: Why AI for Workflows?](13-why-ai-for-workflows.md)
14. [Context Engineering with ChatGPT](14-context-engineering-with-chatgpt.md)
15. [AI Copilot in Kestra](15-ai-copilot-in-kestra.md)
16. [Bonus: Retrieval Augmented Generation (RAG)](16-retrieval-augmented-generation.md)
17. [Bonus: Deploy to the Cloud (Optional)](17-deploy-to-the-cloud.md)
18. [Additional Resources](18-additional-resources.md)

## Homework


## Sections

### 2.1 Introduction to Workflow Orchestration
In this section, you’ll learn the foundations of workflow orchestration, its importance, and how Kestra fits into the orchestration landscape.

### 2.2 Getting Started with Kestra
In this section, you'll learn how to install Kestra, as well as the key concepts required to build your first workflow. Once our first workflow is built, we can extend this further by executing a Python script inside of a workflow. 

You will:
1. Install Kestra using Docker Compose
2. Learn the concepts of Kestra to build your first workflow
3. Execute a Python script inside of a Kestra Flow

### 2.3 Hands-On Coding Project: Build Data Pipelines with Kestra
Next, we're gonna build ETL pipelines for Yellow and Green Taxi data from NYC’s Taxi and Limousine Commission (TLC). You will:
1. Extract data from [CSV files](https://github.com/DataTalksClub/nyc-tlc-data/releases).
2. Load it into Postgres or Google Cloud (GCS + BigQuery).
3. Explore scheduling and backfilling workflows.

### 2.4 ELT Pipelines in Kestra: Google Cloud Platform
Now that you've learned how to build ETL pipelines locally using Postgres, we are ready to move to the cloud. In this section, we'll load the same Yellow and Green Taxi data to Google Cloud Platform (GCP) using: 
1. Google Cloud Storage (GCS) as a data lake  
2. BigQuery as a data warehouse.

### 2.5 Using AI for Data Engineering in Kestra
This section builds on what you learned earlier in Module 2 to show you how AI can speed up workflow development.

By the end of this section, you will:
- Understand why context engineering matters when collaborating with LLMs
- Use AI Copilot to build Kestra flows faster
- Use Retrieval Augmented Generation (RAG) in data pipelines

#### Prerequisites
- Completion of earlier sections in Module 2 (Workflow Orchestration with Kestra)
- Kestra running locally
- Google Cloud account with access to Gemini API (there's a generous free tier!)

---

### 2.6 Bonus: Deploy to the Cloud (Optional)
Now that we've got all our pipelines working and we know how to quickly create new flows with Kestra's AI Copilot, we can deploy Kestra to the cloud so it can continue to orchestrate our scheduled pipelines. 

In this bonus section, we'll cover how you can deploy Kestra on Google Cloud and automatically sync your workflows from a Git repository.

Note: When committing your workflows to Kestra, make sure your workflow doesn't contain any sensitive information. You can use [Secrets](https://go.kestra.io/de-zoomcamp/secret) and the [KV Store](https://go.kestra.io/de-zoomcamp/kv-store) to keep sensitive data out of your workflow logic.

##### Resources
- [Install Kestra on Google Cloud](https://go.kestra.io/de-zoomcamp/gcp-install)
- [Moving from Development to Production](https://go.kestra.io/de-zoomcamp/dev-to-prod)
- [Using Git in Kestra](https://go.kestra.io/de-zoomcamp/git)
- [Deploy Flows with GitHub Actions](https://go.kestra.io/de-zoomcamp/deploy-github-actions)

### 2.7 Additional Resources
- Check [Kestra Docs](https://go.kestra.io/de-zoomcamp/docs)
- Explore our [Blueprints](https://go.kestra.io/de-zoomcamp/blueprints) library
- Browse over 600 [plugins](https://go.kestra.io/de-zoomcamp/plugins) available in Kestra
- Give us a star on [GitHub](https://go.kestra.io/de-zoomcamp/github)
- Join our [Slack community](https://go.kestra.io/de-zoomcamp/slack) if you have any questions
- Find all the videos in this [YouTube Playlist](https://go.kestra.io/de-zoomcamp/yt-playlist)


#### Troubleshooting tips
If you face any issues with Kestra flows in Module 2, make sure to use the following Docker images/ports:
- `image: kestra/kestra:v1.1` - pin your Kestra Docker image to this version so we can ensure reproducibility; do NOT use `kestra/kestra:develop` as this is a bleeding-edge development version that might contain bugs
- `postgres:18` — make sure to pin your Postgres image to version 18
- If you run `pgAdmin` or something else on port 8080, you can adjust Kestra `docker-compose` to use a different port, e.g. change port mapping to 18080 instead of 8080, and then access Kestra UI in your browser from http://localhost:18080/ instead of from http://localhost:8080/

If you are still facing any issues, stop and remove your existing Kestra + Postgres containers and start them again using `docker-compose up -d`. If this doesn't help, post your question on the DataTalksClub Slack or on Kestra's Slack http://kestra.io/slack.

If you encounter similar errors to:

BigQueryError{reason=invalid, location=null,
message=Error while reading table: kestra-sandbox.zooomcamp.yellow_tripdata_2020_01,
error message: CSV table references column position 17, but line contains only 14 columns.;
line_number: 2103925 byte_offset_to_start_of_line: 194863028
column_index: 17 column_name: "congestion_surcharge" column_type: NUMERIC
File: gs://anna-geller/yellow_tripdata_2020-01.csv}

text
It means that the CSV file you're trying to load into BigQuery has a mismatch in the number of columns between the external source table (i.e. file in GCS) and the destination table in BigQuery. This can happen when for due to network/transfer issues, the file is not fully downloaded from GitHub or not correctly uploaded to GCS. The error suggests schema issues but that's not the case. Simply rerun the entire execution including redownloading the CSV file and reuploading it to GCS. This should resolve the issue.

---

## Companion files

* [`docker-compose.yml`](docker-compose.yml) — Kestra, Postgres and pgAdmin
* [`flows/`](flows/) — every Kestra flow used in this module

# Community notes

Did you take notes? You can share them by creating a PR to this file! 

* Add your notes above this line

---

---

## File: 02-workflow-orchestration/01-what-is-workflow-orchestration.md

---
video_url: https://www.youtube.com/watch?v=-JLnp-iLins
prev_url: ../01-docker-terraform/13-gcp-overview.md
next_url: 02-what-is-kestra.md
---
# What is Workflow Orchestration?

Think of a music orchestra. There's a variety of different instruments. Some more than others, all with different roles when it comes to playing music. To make sure they all come together at the right time, they follow a conductor who helps the orchestra to play together.

Now replace the instruments with tools and the conductor with an orchestrator. We often have multiple tools and platforms that we need to work together. Sometimes on a routine schedule, other times based on events that happen. That's where the orchestrator comes in to help all of these tools work together.

A workflow orchestrator might do the following tasks:
- Run workflows which contain a number of predefined steps
- Monitor and log errors, as well as taking a number of extra steps when they occur
- Automatically run workflows based on schedules and events

In data engineering, you often need to move data from one place, to another, sometimes with some modifications made to the data in the middle. This is where a workflow orchestrator can help out by managing these steps, while giving us visibility into it at the same time.


In this module, we're going to build our own data pipeline using ETL (Extract, Transform Load) with Kestra at the core of the operation, but first we need to understand a bit more about how Kestra works before we can get building!

---

## File: 02-workflow-orchestration/02-what-is-kestra.md

---
video_url: https://www.youtube.com/watch?v=ZvVN_NmB_1s
prev_url: 01-what-is-workflow-orchestration.md
next_url: 03-installing-kestra.md
---
# What is Kestra?

Kestra is an open-source, infinitely-scalable orchestration platform that enables all engineers to manage business-critical workflows.

Kestra is a great choice for workflow orchestration:
- Build with Flow code (YAML), No-code or with the AI Copilot - flexibility in how you build your workflows
- 1000+ Plugins - integrate with all the tools you use
- Support for any programming language - pick the right tool for the job
- Schedule or Event Based Triggers - have your workflows respond to data


## Resources
- [Quickstart Guide](https://go.kestra.io/de-zoomcamp/quickstart)
- [What is an Orchestrator?](https://go.kestra.io/de-zoomcamp/what-is-an-orchestrator)

---

## File: 02-workflow-orchestration/03-installing-kestra.md

---
video_url: https://www.youtube.com/watch?v=wgPxC4UjoLM
prev_url: 02-what-is-kestra.md
next_url: 04-kestra-concepts.md
---
# Installing Kestra

To install Kestra, we are going to use Docker Compose. We already have a Postgres database set up, along with pgAdmin from Module 1. We can continue to use these with Kestra but we'll need to make a few modifications to our Docker Compose file.

Use [this example Docker Compose file](docker-compose.yml) to correctly add the 2 new services and set up the volumes correctly.

Add information about setting a username and password.

We'll set up Kestra using Docker Compose containing one container for the Kestra server and another for the Postgres database:


```bash
cd 02-workflow-orchestration
docker compose up -d

Note: Check that pgAdmin isn't running on the same ports as Kestra. If so, check out the FAQ at the bottom of the README.

Once the container starts, you can access the Kestra UI at http://localhost:8080.

To shut down Kestra, go to the same directory and run the following command:

bash
docker compose down

Add Flows to Kestra

Flows can be added to Kestra by copying and pasting the YAML directly into the editor, or by adding via Kestra's API. See below for adding programmatically.

Add Flows to Kestra programmatically

If you prefer to add flows programmatically using Kestra's API, run the following commands:

bash
# Import all flows: assuming username [email protected] and password Admin1234! (adjust to match your username and password)
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/01_hello_world.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/02_python.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/03_getting_started_data_pipeline.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/04_postgres_taxi.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/05_postgres_taxi_scheduled.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/06_gcp_kv.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/07_gcp_setup.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/08_gcp_taxi.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/09_gcp_taxi_scheduled.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/10_chat_without_rag.yaml
curl -X POST -u '[email protected]:Admin1234!' http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/11_chat_with_rag.yaml

Resources

File: 02-workflow-orchestration/04-kestra-concepts.md

video_url: https://www.youtube.com/watch?v=MNOKVx8780E prev_url: 03-installing-kestra.md next_url: 05-orchestrate-python-code.md

Kestra Concepts

To start building workflows in Kestra, we need to understand a number of concepts.

  • Flow - a container for tasks and their orchestration logic.
  • Tasks - the steps within a flow.
  • Inputs - dynamic values passed to the flow at runtime.
  • Outputs - pass data between tasks and flows.
  • Triggers - mechanism that automatically starts the execution of a flow.
  • Execution - a single run of a flow with a specific state.
  • Variables - key–value pairs that let you reuse values across tasks.
  • Plugin Defaults - default values applied to every task of a given type within one or more flows.
  • Concurrency - control how many executions of a flow can run at the same time.

While there are more concepts used for building powerful workflows, these are the ones we're going to use to build our data pipelines.

The flow 01_hello_world.yaml showcases all of these concepts inside of one workflow:

  • The flow has 5 tasks: 3 log tasks and a sleep task
  • The flow takes an input called name.
  • There is a variable that takes the name input to generate a full welcome message.
  • An output is generated from the return task and is logged in a later log task.
  • There is a trigger to execute this flow every day at 10am.
  • Plugin Defaults are used to make both log tasks send their messages as ERROR level.
  • We have a concurrency limit of 2 executions. Any further ones made while 2 are running will fail.

Resources

File: 02-workflow-orchestration/05-orchestrate-python-code.md

video_url: https://www.youtube.com/watch?v=VAHm0R_XjqI prev_url: 04-kestra-concepts.md next_url: 06-getting-started-pipeline.md

Orchestrate Python Code

Now that we've built our first workflow, we can take it a step further by adding Python code into our flow. In Kestra, we can run Python code from a dedicated file or write it directly inside of our workflow.

While Kestra has a huge variety of plugins available for building your workflows, you also have the option to write your own code and have Kestra execute that based on schedules or events. This means you can pick the right tools for your pipelines, rather than the ones you're limited to.

In our example Python workflow, 02_python.yaml, our code fetches the number of Docker image pulls from DockerHub and returns it as an output to Kestra. This is useful as we can access this output with other tasks, even though it was generated inside of our Python script.

Resources

File: 02-workflow-orchestration/06-getting-started-pipeline.md

video_url: https://www.youtube.com/watch?v=-KmwrCqRhic prev_url: 05-orchestrate-python-code.md next_url: 07-load-taxi-data-to-postgres.md

Getting Started Pipeline

This introductory flow is added just to demonstrate a simple data pipeline which extracts data via HTTP REST API, transforms that data in Python and then queries it using DuckDB. For this stage, a new separate Postgres database is created for the exercises.

mermaid
graph LR
  Extract[Extract Data via HTTP REST API] --> Transform[Transform Data in Python]
  Transform --> Query[Query Data with DuckDB]

Add the flow 03_getting_started_data_pipeline.yaml from the UI if you haven't already and execute it to see the results. Inspect the Gantt and Logs tabs to understand the flow execution.

Resources

File: 02-workflow-orchestration/07-load-taxi-data-to-postgres.md

video_url: https://www.youtube.com/watch?v=Z9ZmmwtXDcU prev_url: 06-getting-started-pipeline.md next_url: 08-scheduling-and-backfills.md

Local DB: Load Taxi Data to Postgres

Before we start loading data to GCP, we'll first play with the Yellow and Green Taxi data using a local Postgres database running in a Docker container. We will use the same database from Module 1 which should be in the same Docker Compose file as Kestra.

The flow will extract CSV data partitioned by year and month, create tables, load data to the monthly table, and finally merge the data to the final destination table.

mermaid
graph LR
  Start[Select Year & Month] --> SetLabel[Set Labels]
  SetLabel --> Extract[Extract CSV Data]
  Extract -->|Taxi=Yellow| YellowFinalTable[Create Yellow Final Table]:::yellow
  Extract -->|Taxi=Green| GreenFinalTable[Create Green Final Table]:::green
  YellowFinalTable --> YellowMonthlyTable[Create Yellow Monthly Table]:::yellow
  GreenFinalTable --> GreenMonthlyTable[Create Green Monthly Table]:::green
  YellowMonthlyTable --> YellowCopyIn[Load Data to Monthly Table]:::yellow
  GreenMonthlyTable --> GreenCopyIn[Load Data to Monthly Table]:::green
  YellowCopyIn --> YellowMerge[Merge Yellow Data]:::yellow
  GreenCopyIn --> GreenMerge[Merge Green Data]:::green

  classDef yellow fill:#FFD700,stroke:#000,stroke-width:1px,color:#000;
  classDef green fill:#32CD32,stroke:#000,stroke-width:1px,color:#000;

The flow code: 04_postgres_taxi.yaml.

[!NOTE]
The NYC Taxi and Limousine Commission (TLC) Trip Record Data provided on the nyc.gov website is currently available only in a Parquet format, but this is NOT the dataset we're going to use in this course. For the purpose of this course, we'll use the CSV files available here on GitHub. This is because the Parquet format can be challenging to understand by newcomers, and we want to make the course as accessible as possible — the CSV format can be easily introspected using tools like Excel or Google Sheets, or even a simple text editor.

Resources

File: 02-workflow-orchestration/08-scheduling-and-backfills.md

video_url: https://www.youtube.com/watch?v=1pu_C_oOAMA prev_url: 07-load-taxi-data-to-postgres.md next_url: 09-etl-vs-elt.md

Local DB: Learn Scheduling and Backfills

We can now schedule the same pipeline shown above to run daily at 9 AM UTC. We'll also demonstrate how to backfill the data pipeline to run on historical data.

Note: given the large dataset, we'll backfill only data for the green taxi dataset for the year 2019.

The flow code: 05_postgres_taxi_scheduled.yaml.

File: 02-workflow-orchestration/09-etl-vs-elt.md

video_url: https://www.youtube.com/watch?v=E04yurp1tSU prev_url: 08-scheduling-and-backfills.md next_url: 10-setup-google-cloud-platform.md

ETL vs ELT

In 2.3, we made a ETL pipeline inside of Kestra:

  • Extract: Firstly, we extract the dataset from GitHub
  • Transform: Next, we transform it with Python
  • Load: Finally, we load it into our Postgres database

While this is very standard across the industry, sometimes it makes sense to change the order when working with the cloud. If you're working with a large dataset, like the Yellow Taxi data, there can be benefits to extracting and loading straight into a data warehouse, and then performing transformations directly in the data warehouse. When working with BigQuery, we will use ELT:

  • Extract: Firstly, we extract the dataset from GitHub
  • Load: Next, we load this dataset (in this case, a csv file) into a data lake (Google Cloud Storage)
  • Transform: Finally, we can create a table inside of our data warehouse (BigQuery) which uses the data from our data lake to perform our transformations.

The reason for loading into the data warehouse before transforming means we can utilize the cloud's performance benefits for transforming large datasets. What might take a lot longer for a local machine, can take a fraction of the time in the cloud.

Over the next few videos, we'll look at setting up BigQuery and transforming the Yellow Taxi dataset.

Resources

File: 02-workflow-orchestration/10-setup-google-cloud-platform.md

video_url: https://www.youtube.com/watch?v=TLGFAOHpOYM prev_url: 09-etl-vs-elt.md next_url: 11-load-taxi-data-to-bigquery.md

Setup Google Cloud Platform (GCP)

Before we start loading data to GCP, we need to set up the Google Cloud Platform.

First, adjust the following flow 06_gcp_kv.yaml to include your service account, GCP project ID, BigQuery dataset and GCS bucket name (along with their location) as KV Store values:

  • GCP_PROJECT_ID
  • GCP_LOCATION
  • GCP_BUCKET_NAME
  • GCP_DATASET.

Create GCP Resources

If you haven't already created the GCS bucket and BigQuery dataset in the first week of the course, you can use this flow to create them: 07_gcp_setup.yaml.

[!WARNING]
The GCP_CREDS service account contains sensitive information. Ensure you keep it secure and do not commit it to Git. Keep it as secure as your passwords.

Resources

File: 02-workflow-orchestration/11-load-taxi-data-to-bigquery.md

video_url: https://www.youtube.com/watch?v=52u9X_bfTAo prev_url: 10-setup-google-cloud-platform.md next_url: 12-schedule-and-backfill-full-dataset.md

GCP Workflow: Load Taxi Data to BigQuery

Now that Google Cloud is set up with a storage bucket, we can start the ELT process.

mermaid
graph LR
  SetLabel[Set Labels] --> Extract[Extract CSV Data]
  Extract --> UploadToGCS[Upload Data to GCS]
  UploadToGCS -->|Taxi=Yellow| BQYellowTripdata[Main Yellow Tripdata Table]:::yellow
  UploadToGCS -->|Taxi=Green| BQGreenTripdata[Main Green Tripdata Table]:::green
  BQYellowTripdata --> BQYellowTableExt[External Table]:::yellow
  BQGreenTripdata --> BQGreenTableExt[External Table]:::green
  BQYellowTableExt --> BQYellowTableTmp[Monthly Table]:::yellow
  BQGreenTableExt --> BQGreenTableTmp[Monthly Table]:::green
  BQYellowTableTmp --> BQYellowMerge[Merge to Main Table]:::yellow
  BQGreenTableTmp --> BQGreenMerge[Merge to Main Table]:::green
  BQYellowMerge --> PurgeFiles[Purge Files]
  BQGreenMerge --> PurgeFiles[Purge Files]

  classDef yellow fill:#FFD700,stroke:#000,stroke-width:1px,color:#000
  classDef green fill:#32CD32,stroke:#000,stroke-width:1px,color:#000

The flow code: 08_gcp_taxi.yaml.

File: 02-workflow-orchestration/12-schedule-and-backfill-full-dataset.md

video_url: https://www.youtube.com/watch?v=b-6KhfWfk2M prev_url: 11-load-taxi-data-to-bigquery.md next_url: 13-why-ai-for-workflows.md

GCP Workflow: Schedule and Backfill Full Dataset

We can now schedule the same pipeline shown above to run daily at 9 AM UTC for the green dataset and at 10 AM UTC for the yellow dataset. You can backfill historical data directly from the Kestra UI.

Since we now process data in a cloud environment with infinitely scalable storage and compute, we can backfill the entire dataset for both the yellow and green taxi data without the risk of running out of resources on our local machine.

The flow code: 09_gcp_taxi_scheduled.yaml.

File: 02-workflow-orchestration/13-why-ai-for-workflows.md

video_url: https://www.youtube.com/watch?v=GHPtRDAv044 prev_url: 12-schedule-and-backfill-full-dataset.md next_url: 14-context-engineering-with-chatgpt.md

Introduction: Why AI for Workflows?

As data engineers, we spend significant time writing boilerplate code, searching documentation, and structuring data pipelines. AI tools can help us:

  • Generate workflows faster: Describe what you want to accomplish in natural language instead of writing YAML from scratch
  • Avoid errors: Get syntax-correct, up-to-date workflow code that follows best practices

However, AI is only as good as the context we provide. This section teaches you how to engineer that context for reliable, production-ready data workflows.

AI can accelerate workflow creation, but context, validation, and human review remain the path to reliable deployment.

File: 02-workflow-orchestration/14-context-engineering-with-chatgpt.md

video_url: https://www.youtube.com/watch?v=LmnfjGKwnVU prev_url: 13-why-ai-for-workflows.md next_url: 15-ai-copilot-in-kestra.md

Context Engineering with ChatGPT

The comparison is about grounding: relevant versions, documentation, and examples turn a vague prompt into a reviewable workflow.

Let's start by seeing what happens when AI lacks proper context.

Experiment: ChatGPT Without Context

  1. Open ChatGPT in a private browser window (to avoid any existing chat context): https://chatgpt.com

  2. Enter this prompt:

    text
    Create a Kestra flow that loads NYC taxi data from a CSV file to BigQuery. The flow should extract data, upload to GCS, and load to BigQuery.

  3. Observe the results:

    • ChatGPT will generate a Kestra flow, but it likely contains:
      • Outdated plugin syntax e.g., old task types that have been renamed
      • Incorrect property names e.g., properties that don't exist in current versions
      • Hallucinated features e.g., tasks, triggers or properties that never existed

Why Does This Happen?

Large Language Models (LLMs) like GPT models from OpenAI are trained on data up to a specific point in time (knowledge cutoff). They don't automatically know about:

  • Software updates and new releases
  • Renamed plugins or changed APIs

This is the fundamental challenge of using AI: the model can only work with information it has access to.

Key Learning: Context is Everything

Without proper context:

  • ❌ Generic AI assistants hallucinate outdated or incorrect code
  • ❌ You can't trust the output for production use

With proper context:

  • ✅ AI generates accurate, current, production-ready code
  • ✅ You can iterate faster by letting AI generate boilerplate workflow code

In the next section, we'll see how Kestra's AI Copilot solves this problem.

File: 02-workflow-orchestration/15-ai-copilot-in-kestra.md

video_url: https://www.youtube.com/watch?v=3IbjHfC8bMg prev_url: 14-context-engineering-with-chatgpt.md next_url: 16-retrieval-augmented-generation.md

AI Copilot in Kestra

Kestra's AI Copilot is specifically designed to generate and modify Kestra flows with full context about the latest plugins, workflow syntax, and best practices.

Copilot’s advantage is contextual assistance: it drafts within Kestra’s flow model while keeping review and execution explicit.

Setup AI Copilot

Before using AI Copilot, you need to configure Gemini API access in your Kestra instance.

Step 1: Get Your Gemini API Key

  1. Visit Google AI Studio: https://aistudio.google.com/app/apikey
  2. Sign in with your Google account
  3. Click "Create API Key"
  4. Copy the generated key (keep it secure!)

[!WARNING]
Never commit API keys to Git. Always use environment variables or Kestra's KV Store.

Step 2: Configure Kestra AI Copilot

Add the following to your Kestra configuration. You can do this by modifying your docker-compose.yml file from 2.2:

yaml
services:
  kestra:
    environment:
      KESTRA_CONFIGURATION: |
        kestra:
          ai:
            type: gemini
            gemini:
              model-name: gemini-2.5-flash
              api-key: ${GEMINI_API_KEY}

Then restart Kestra:

bash
cd 02-workflow-orchestration/docker
export GEMINI_API_KEY="your-api-key-here"
docker compose up -d

Exercise: ChatGPT vs AI Copilot Comparison

Objective: Learn why context engineering matters.

  1. Open Kestra UI at http://localhost:8080
  2. Create a new flow and open the Code editor panel
  3. Click the AI Copilot button (sparkle icon ✨) in the top-right corner
  4. Enter the same exact prompt we used with ChatGPT:
    text
    Create a Kestra flow that loads NYC taxi data from a CSV file to BigQuery. The flow should extract data, upload to GCS, and load to BigQuery.
  5. Compare the outputs:
    • ✅ Copilot generates executable, working YAML
    • ✅ Copilot uses correct plugin types and properties
    • ✅ Copilot follows current Kestra best practices

Key Learning: Context matters! AI Copilot has access to current Kestra documentation, generating Kestra flows better than a generic ChatGPT assistant.

File: 02-workflow-orchestration/16-retrieval-augmented-generation.md

video_url: https://www.youtube.com/watch?v=XuPDQ1UcNyI prev_url: 15-ai-copilot-in-kestra.md next_url: 17-deploy-to-the-cloud.md

Bonus: Retrieval Augmented Generation (RAG)

To further learn how to provide context to your prompts, this bonus section demonstrates how to use RAG.

What is RAG?

RAG (Retrieval Augmented Generation) is a technique that:

  1. Retrieves relevant information from your data sources
  2. Augments the AI prompt with this context
  3. Generates a response grounded in real data

This solves the hallucination problem by ensuring the AI has access to current, accurate information at query time.

RAG adds a retrieval path between the question and the model so answers can be grounded in current source material.

How RAG Works in Kestra

mermaid
graph LR
    A[Ask AI] --> B[Fetch Docs]
    B --> C[Create Embeddings]
    C --> D[Find Similar Content]
    D --> E[Add Context to Prompt]
    E --> F[LLM Answer]

The Process:

  1. Ingest documents: Load documentation, release notes, or other data sources
  2. Create embeddings: Convert text into vector representations using an LLM
  3. Store embeddings: Save vectors in Kestra's KV Store (or a vector database)
  4. Query with context: When you ask a question, retrieve relevant embeddings and include them in the prompt
  5. Generate response: The LLM has real context and provides accurate answers

Exercise: Retrieval With vs Without Context

Objective: Understand how RAG eliminates hallucinations by grounding LLM responses in real data.

Part A: Without RAG

  1. Navigate to the 10_chat_without_rag.yaml flow in your Kestra UI
  2. Click Execute
  3. Wait for the execution to complete
  4. Open the Logs tab
  5. Read the output - notice how the response about "Kestra 1.1 features" is:
    • Vague or generic
    • Potentially incorrect
    • Missing specific details
    • Based only on the model's training data (which may be outdated)

Part B: With RAG

  1. Navigate to the 11_chat_with_rag.yaml flow
  2. Click Execute
  3. Watch the execution:
    • First task: Ingests Kestra 1.1 release documentation, creates embeddings and stores them
    • Second task: Prompts LLM with context retrieved from stored embeddings
  4. Open the Logs tab
  5. Compare this output with the previous one - notice how it's:
    • ✅ Specific and detailed
    • ✅ Accurate with real features from the release
    • ✅ Grounded in actual documentation

Key Learning: RAG (Retrieval Augmented Generation) grounds AI responses in current documentation, eliminating hallucinations and providing accurate, context-aware answers.

RAG Best Practices

  1. Keep documents updated: Regularly re-ingest to ensure current information
  2. Chunk appropriately: Break large documents into meaningful chunks
  3. Test retrieval quality: Verify that the right documents are retrieved

Additional AI Resources

Kestra Documentation:

Kestra Plugin Documentation:

External Documentation:

File: 02-workflow-orchestration/17-deploy-to-the-cloud.md

prev_url: 16-retrieval-augmented-generation.md next_url: 18-additional-resources.md

Bonus: Deploy to the Cloud (Optional)

Now that we've got all our pipelines working and we know how to quickly create new flows with Kestra's AI Copilot, we can deploy Kestra to the cloud so it can continue to orchestrate our scheduled pipelines.

In this bonus section, we'll cover how you can deploy Kestra on Google Cloud and automatically sync your workflows from a Git repository.

Cloud deployment connects reviewed workflow code to scheduled execution while keeping credentials in a separate secret store.

Note: When committing your workflows to Kestra, make sure your workflow doesn't contain any sensitive information. You can use Secrets and the KV Store to keep sensitive data out of your workflow logic.

Resources

File: 02-workflow-orchestration/18-additional-resources.md

prev_url: 17-deploy-to-the-cloud.md next_url: ../03-data-warehouse/01-data-warehouse-and-bigquery.md

Additional Resources

The resource set is a map for continuing beyond the lesson: reference material, examples, tools, and help channels serve different needs.

Troubleshooting tips

If you face any issues with Kestra flows in Module 2, make sure to use the following Docker images/ports:

  • image: kestra/kestra:v1.1 - pin your Kestra Docker image to this version so we can ensure reproducibility; do NOT use kestra/kestra:develop as this is a bleeding-edge development version that might contain bugs
  • postgres:18 — make sure to pin your Postgres image to version 18
  • If you run pgAdmin or something else on port 8080, you can adjust Kestra docker-compose to use a different port, e.g. change port mapping to 18080 instead of 8080, and then access Kestra UI in your browser from http://localhost:18080/ instead of from http://localhost:8080/

If you are still facing any issues, stop and remove your existing Kestra + Postgres containers and start them again using docker-compose up -d. If this doesn't help, post your question on the DataTalksClub Slack or on Kestra's Slack http://kestra.io/slack.

If you encounter similar errors to:

text
BigQueryError{reason=invalid, location=null,
message=Error while reading table: kestra-sandbox.zooomcamp.yellow_tripdata_2020_01,
error message: CSV table references column position 17, but line contains only 14 columns.;
line_number: 2103925 byte_offset_to_start_of_line: 194863028
column_index: 17 column_name: "congestion_surcharge" column_type: NUMERIC
File: gs://anna-geller/yellow_tripdata_2020-01.csv}

It means that the CSV file you're trying to load into BigQuery has a mismatch in the number of columns between the external source table (i.e. file in GCS) and the destination table in BigQuery. This can happen when for due to network/transfer issues, the file is not fully downloaded from GitHub or not correctly uploaded to GCS. The error suggests schema issues but that's not the case. Simply rerun the entire execution including redownloading the CSV file and reuploading it to GCS. This should resolve the issue.

File: 02-workflow-orchestration/docker-compose.yml

volumes:
ny_taxi_postgres_data:
driver: local
kestra_postgres_data:
driver: local
kestra_data:
driver: local
kestra_tmp:
driver: local

services:
pgdatabase:
image: postgres:18
environment:
POSTGRES_USER: root
POSTGRES_PASSWORD: root
POSTGRES_DB: ny_taxi
ports:
- "5432:5432"
volumes:
- ny_taxi_postgres_data:/var/lib/postgresql
depends_on:
kestra:
condition: service_started

pgadmin:
image: dpage/pgadmin4
environment:
- PGADMIN_DEFAULT_EMAIL=[email protected]
- PGADMIN_DEFAULT_PASSWORD=root
ports:
- "8085:80"
depends_on:
pgdatabase:
condition: service_started

kestra_postgres:
image: postgres:18
volumes:
- kestra_postgres_data:/var/lib/postgresql
environment:
POSTGRES_DB: kestra
POSTGRES_USER: kestra
POSTGRES_PASSWORD: k3str4
healthcheck:
test: ["CMD-SHELL", "pg_isready -d $${POSTGRES_DB} -U $${POSTGRES_USER}"]
interval: 30s
timeout: 10s
retries: 10

kestra:
image: kestra/kestra:v1.1
pull_policy: always
# Note that this setup with a root user is intended for development purpose.
# Our base image runs without root, but the Docker Compose implementation needs root to access the Docker socket
# To run Kestra in a rootless mode in production, see: https://kestra.io/docs/installation/podman-compose
user: "root"
command: server standalone
volumes:
- kestra_data:/app/storage
- /var/run/docker.sock:/var/run/docker.sock
- kestra_tmp:/tmp/kestra-wd
environment:
KESTRA_CONFIGURATION: |
datasources:
postgres:
url: jdbc:postgresql://kestra_postgres:5432/kestra
driverClassName: org.postgresql.Driver
username: kestra
password: k3str4
kestra:
server:
basicAuth:
username: "[email protected]" # it must be a valid email address
password: Admin1234!
type: postgres
storage:
type: local
local:
basePath: "/app/storage"
queue:
type: postgres
tasks:
tmpDir:
path: /tmp/kestra-wd/tmp
url: http://localhost:8080/
ports:
- "8080:8080"
- "8081:8081"
depends_on:
kestra_postgres:
condition: service_started

File: 02-workflow-orchestration/module.yaml

schema_version: 2
content_id: 2e4456c8-9215-461c-ba70-69648f6f99e8
title: "Workflow Orchestration"
summary: "Build scheduled ETL and ELT workflows with Kestra, including backfills and cloud deployment."
units:

  • content_id: 438f15e4-0529-465e-a08a-b8d17a03bcd1
    title: "What is Workflow Orchestration?"
    path: 01-what-is-workflow-orchestration.md
  • content_id: dc00bc6a-b407-4217-a87e-66b6ec4488cb
    title: "What is Kestra?"
    path: 02-what-is-kestra.md
  • content_id: 82b57e73-ee30-412f-a18f-3e21272e9586
    title: "Installing Kestra"
    path: 03-installing-kestra.md
  • content_id: e1b1d6df-852f-4013-b170-3096b449d904
    title: "Kestra Concepts"
    path: 04-kestra-concepts.md
  • content_id: 42dadc02-219c-4136-9d9d-45576ee36bdf
    title: "Orchestrate Python Code"
    path: 05-orchestrate-python-code.md
  • content_id: c43f5d3a-9556-4072-bb62-2082102a8b20
    title: "Getting Started Pipeline"
    path: 06-getting-started-pipeline.md
  • content_id: da7d6b7e-6488-4e09-afaa-43a79afd8141
    title: "Local DB: Load Taxi Data to Postgres"
    path: 07-load-taxi-data-to-postgres.md
  • content_id: 78bc0459-0220-4083-be22-8f81b1585ff7
    title: "Local DB: Learn Scheduling and Backfills"
    path: 08-scheduling-and-backfills.md
  • content_id: 22552137-6d60-40d8-b51d-d936e8c4ae27
    title: "ETL vs ELT"
    path: 09-etl-vs-elt.md
  • content_id: 54beab06-48e7-470d-9ab4-ffa28b9aea7c
    title: "Setup Google Cloud Platform (GCP)"
    path: 10-setup-google-cloud-platform.md
  • content_id: a95e13c2-ec9a-44a5-8e0b-dcea3d7de6e6
    title: "GCP Workflow: Load Taxi Data to BigQuery"
    path: 11-load-taxi-data-to-bigquery.md
  • content_id: 574c9a50-cd8b-4e0c-b678-eb1b1ca9078c
    title: "GCP Workflow: Schedule and Backfill Full Dataset"
    path: 12-schedule-and-backfill-full-dataset.md
  • content_id: f3839f17-fe8b-4802-8d83-c28bca54cff8
    title: "Introduction: Why AI for Workflows?"
    path: 13-why-ai-for-workflows.md
  • content_id: 50278dfb-e12d-4a37-8eae-bd6cb7ff22ba
    title: "Context Engineering with ChatGPT"
    path: 14-context-engineering-with-chatgpt.md
  • content_id: 17bef640-09e7-4e8a-b7c8-95d2162e0e4a
    title: "AI Copilot in Kestra"
    path: 15-ai-copilot-in-kestra.md
  • content_id: d2e82a7d-2c22-4017-8b70-12c3b8c99d0f
    title: "Bonus: Retrieval Augmented Generation (RAG)"
    path: 16-retrieval-augmented-generation.md
  • content_id: c09c8fcd-2c50-4605-9088-aaece3e12945
    title: "Bonus: Deploy to the Cloud (Optional)"
    path: 17-deploy-to-the-cloud.md
  • content_id: a2673bba-259f-4f59-84f8-930abac93828
    title: "Additional Resources"
    path: 18-additional-resources.md

File: 02-workflow-orchestration/flows/01_hello_world.yaml

id: 01_hello_world
namespace: zoomcamp

inputs:

  • id: name
    type: STRING
    defaults: Will

concurrency:
behavior: FAIL
limit: 2

variables:
welcome_message: "Hello, {{ inputs.name }}!"

tasks:

  • id: hello_message
    type: io.kestra.plugin.core.log.Log
    message: "{{ render(vars.welcome_message) }}"

  • id: generate_output
    type: io.kestra.plugin.core.debug.Return
    format: I was generated during this workflow.

  • id: sleep
    type: io.kestra.plugin.core.flow.Sleep
    duration: PT15S

  • id: log_output
    type: io.kestra.plugin.core.log.Log
    message: "This is an output: {{ outputs.generate_output.value }}"

  • id: goodbye_message
    type: io.kestra.plugin.core.log.Log
    message: "Goodbye, {{ inputs.name }}!"

pluginDefaults:

  • type: io.kestra.plugin.core.log.Log
    values:
    level: ERROR

triggers:

  • id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 10 * * *"
    inputs:
    name: Sarah
    disabled: true

File: 02-workflow-orchestration/flows/02_python.yaml

id: 02_python
namespace: zoomcamp
tasks:

  • id: collect_stats
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
    type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:slim
    dependencies:
    • requests
    • kestra
      script: |
      from kestra import Kestra
      import requests
      def get_docker_image_downloads(image_name: str = "kestra/kestra"):
      """Queries the Docker Hub API to get the number of downloads for a specific Docker image."""
      url = f"https://hub.docker.com/v2/repositories/{image_name}/"
      response = requests.get(url)
      data = response.json()
      downloads = data.get('pull_count', 'Not available')
      return downloads
      downloads = get_docker_image_downloads()
      outputs = {
      'downloads': downloads
      }
      Kestra.outputs(outputs)

File: 02-workflow-orchestration/flows/03_getting_started_data_pipeline.yaml

id: 03_getting_started_data_pipeline
namespace: zoomcamp

inputs:

  • id: columns_to_keep
    type: ARRAY
    itemType: STRING
    defaults:
    • brand
    • price

tasks:

  • id: extract
    type: io.kestra.plugin.core.http.Download
    uri: https://dummyjson.com/products

  • id: transform
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-alpine
    inputFiles:
    data.json: "{{outputs.extract.uri}}"
    outputFiles:

    • "*.json"
      env:
      COLUMNS_TO_KEEP: "{{inputs.columns_to_keep}}"
      script: |
      import json
      import os

    columns_to_keep_str = os.getenv("COLUMNS_TO_KEEP")
    columns_to_keep = json.loads(columns_to_keep_str)

    with open("data.json", "r") as file:
    data = json.load(file)

    filtered_data = [
    {column: product.get(column, "N/A") for column in columns_to_keep}
    for product in data["products"]
    ]

    with open("products.json", "w") as file:
    json.dump(filtered_data, file, indent=4)

  • id: query
    type: io.kestra.plugin.jdbc.duckdb.Queries
    inputFiles:
    products.json: "{{outputs.transform.outputFiles['products.json']}}"
    sql: |
    INSTALL json;
    LOAD json;
    SELECT brand, round(avg(price), 2) as avg_price
    FROM read_json_auto('{{workingDir}}/products.json')
    GROUP BY brand
    ORDER BY avg_price DESC;
    fetchType: STORE

File: 02-workflow-orchestration/flows/04_postgres_taxi.yaml

id: 04_postgres_taxi
namespace: zoomcamp
The CSV Data used in the course: https://github.com/DataTalksClub/nyc-tlc-data/releases

inputs:

  • id: taxi
    type: SELECT
    displayName: Select taxi type
    values: [yellow, green]
    defaults: yellow

  • id: year
    type: SELECT
    displayName: Select year
    values: ["2019", "2020"]
    defaults: "2019"

  • id: month
    type: SELECT
    displayName: Select month
    values: ["01", "02", "03", "04", "05", "06", "07", "08", "09", "10", "11", "12"]
    defaults: "01"

variables:
file: "{{inputs.taxi}}tripdata{{inputs.year}}-{{inputs.month}}.csv"
staging_table: "public.{{inputs.taxi}}_tripdata_staging"
table: "public.{{inputs.taxi}}_tripdata"
data: "{{outputs.extract.outputFiles[inputs.taxi ~ 'tripdata' ~ inputs.year ~ '-' ~ inputs.month ~ '.csv']}}"

tasks:

  • id: set_label
    type: io.kestra.plugin.core.execution.Labels
    labels:
    file: "{{render(vars.file)}}"
    taxi: "{{inputs.taxi}}"

  • id: extract
    type: io.kestra.plugin.scripts.shell.Commands
    outputFiles:

  • id: if_yellow_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'yellow'}}"
    then:

    • id: yellow_create_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      tpep_pickup_datetime timestamp,
      tpep_dropoff_datetime timestamp,
      passenger_count integer,
      trip_distance double precision,
      RatecodeID text,
      store_and_fwd_flag text,
      PULocationID text,
      DOLocationID text,
      payment_type integer,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      congestion_surcharge double precision

      );

    • id: yellow_create_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.staging_table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      tpep_pickup_datetime timestamp,
      tpep_dropoff_datetime timestamp,
      passenger_count integer,
      trip_distance double precision,
      RatecodeID text,
      store_and_fwd_flag text,
      PULocationID text,
      DOLocationID text,
      payment_type integer,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      congestion_surcharge double precision

      );

    • id: yellow_truncate_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      TRUNCATE TABLE {{render(vars.staging_table)}};

    • id: yellow_copy_in_to_staging_table
      type: io.kestra.plugin.jdbc.postgresql.CopyIn
      format: CSV
      from: "{{render(vars.data)}}"
      table: "{{render(vars.staging_table)}}"
      header: true
      columns: [VendorID,tpep_pickup_datetime,tpep_dropoff_datetime,passenger_count,trip_distance,RatecodeID,store_and_fwd_flag,PULocationID,DOLocationID,payment_type,fare_amount,extra,mta_tax,tip_amount,tolls_amount,improvement_surcharge,total_amount,congestion_surcharge]

    • id: yellow_add_unique_id_and_filename
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      UPDATE {{render(vars.staging_table)}}
      SET
      unique_row_id = md5(
      COALESCE(CAST(VendorID AS text), '') ||
      COALESCE(CAST(tpep_pickup_datetime AS text), '') ||
      COALESCE(CAST(tpep_dropoff_datetime AS text), '') ||
      COALESCE(PULocationID, '') ||
      COALESCE(DOLocationID, '') ||
      COALESCE(CAST(fare_amount AS text), '') ||
      COALESCE(CAST(trip_distance AS text), '')
      ),
      filename = '{{render(vars.file)}}';

    • id: yellow_merge_data
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      MERGE INTO {{render(vars.table)}} AS T
      USING {{render(vars.staging_table)}} AS S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (
      unique_row_id, filename, VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,
      passenger_count, trip_distance, RatecodeID, store_and_fwd_flag, PULocationID,
      DOLocationID, payment_type, fare_amount, extra, mta_tax, tip_amount, tolls_amount,
      improvement_surcharge, total_amount, congestion_surcharge
      )
      VALUES (
      S.unique_row_id, S.filename, S.VendorID, S.tpep_pickup_datetime, S.tpep_dropoff_datetime,
      S.passenger_count, S.trip_distance, S.RatecodeID, S.store_and_fwd_flag, S.PULocationID,
      S.DOLocationID, S.payment_type, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount,
      S.improvement_surcharge, S.total_amount, S.congestion_surcharge
      );

  • id: if_green_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'green'}}"
    then:

    • id: green_create_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      lpep_pickup_datetime timestamp,
      lpep_dropoff_datetime timestamp,
      store_and_fwd_flag text,
      RatecodeID text,
      PULocationID text,
      DOLocationID text,
      passenger_count integer,
      trip_distance double precision,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      ehail_fee double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      payment_type integer,
      trip_type integer,
      congestion_surcharge double precision

      );

    • id: green_create_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.staging_table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      lpep_pickup_datetime timestamp,
      lpep_dropoff_datetime timestamp,
      store_and_fwd_flag text,
      RatecodeID text,
      PULocationID text,
      DOLocationID text,
      passenger_count integer,
      trip_distance double precision,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      ehail_fee double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      payment_type integer,
      trip_type integer,
      congestion_surcharge double precision

      );

    • id: green_truncate_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      TRUNCATE TABLE {{render(vars.staging_table)}};

    • id: green_copy_in_to_staging_table
      type: io.kestra.plugin.jdbc.postgresql.CopyIn
      format: CSV
      from: "{{render(vars.data)}}"
      table: "{{render(vars.staging_table)}}"
      header: true
      columns: [VendorID,lpep_pickup_datetime,lpep_dropoff_datetime,store_and_fwd_flag,RatecodeID,PULocationID,DOLocationID,passenger_count,trip_distance,fare_amount,extra,mta_tax,tip_amount,tolls_amount,ehail_fee,improvement_surcharge,total_amount,payment_type,trip_type,congestion_surcharge]

    • id: green_add_unique_id_and_filename
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      UPDATE {{render(vars.staging_table)}}
      SET
      unique_row_id = md5(
      COALESCE(CAST(VendorID AS text), '') ||
      COALESCE(CAST(lpep_pickup_datetime AS text), '') ||
      COALESCE(CAST(lpep_dropoff_datetime AS text), '') ||
      COALESCE(PULocationID, '') ||
      COALESCE(DOLocationID, '') ||
      COALESCE(CAST(fare_amount AS text), '') ||
      COALESCE(CAST(trip_distance AS text), '')
      ),
      filename = '{{render(vars.file)}}';

    • id: green_merge_data
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      MERGE INTO {{render(vars.table)}} AS T
      USING {{render(vars.staging_table)}} AS S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (
      unique_row_id, filename, VendorID, lpep_pickup_datetime, lpep_dropoff_datetime,
      store_and_fwd_flag, RatecodeID, PULocationID, DOLocationID, passenger_count,
      trip_distance, fare_amount, extra, mta_tax, tip_amount, tolls_amount, ehail_fee,
      improvement_surcharge, total_amount, payment_type, trip_type, congestion_surcharge
      )
      VALUES (
      S.unique_row_id, S.filename, S.VendorID, S.lpep_pickup_datetime, S.lpep_dropoff_datetime,
      S.store_and_fwd_flag, S.RatecodeID, S.PULocationID, S.DOLocationID, S.passenger_count,
      S.trip_distance, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount, S.ehail_fee,
      S.improvement_surcharge, S.total_amount, S.payment_type, S.trip_type, S.congestion_surcharge
      );

  • id: purge_files
    type: io.kestra.plugin.core.storage.PurgeCurrentExecutionFiles
    pluginDefaults:

  • type: io.kestra.plugin.jdbc.postgresql
    values:
    url: jdbc:postgresql://pgdatabase:5432/ny_taxi
    username: root
    password: root

File: 02-workflow-orchestration/flows/05_postgres_taxi_scheduled.yaml

id: 05_postgres_taxi_scheduled
namespace: zoomcamp
Best to add a label backfill:true from the UI to track executions created via a backfill.
CSV data used here comes from: https://github.com/DataTalksClub/nyc-tlc-data/releases

concurrency:
limit: 1

inputs:

  • id: taxi
    type: SELECT
    displayName: Select taxi type
    values: [yellow, green]
    defaults: yellow

variables:
file: "{{inputs.taxi}}tripdata{{trigger.date | date('yyyy-MM')}}.csv"
staging_table: "public.{{inputs.taxi}}_tripdata_staging"
table: "public.{{inputs.taxi}}_tripdata"
data: "{{outputs.extract.outputFiles[inputs.taxi ~ 'tripdata' ~ (trigger.date | date('yyyy-MM')) ~ '.csv']}}"

tasks:

  • id: set_label
    type: io.kestra.plugin.core.execution.Labels
    labels:
    file: "{{render(vars.file)}}"
    taxi: "{{inputs.taxi}}"

  • id: extract
    type: io.kestra.plugin.scripts.shell.Commands
    outputFiles:

  • id: if_yellow_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'yellow'}}"
    then:

    • id: yellow_create_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      tpep_pickup_datetime timestamp,
      tpep_dropoff_datetime timestamp,
      passenger_count integer,
      trip_distance double precision,
      RatecodeID text,
      store_and_fwd_flag text,
      PULocationID text,
      DOLocationID text,
      payment_type integer,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      congestion_surcharge double precision

      );

    • id: yellow_create_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.staging_table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      tpep_pickup_datetime timestamp,
      tpep_dropoff_datetime timestamp,
      passenger_count integer,
      trip_distance double precision,
      RatecodeID text,
      store_and_fwd_flag text,
      PULocationID text,
      DOLocationID text,
      payment_type integer,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      congestion_surcharge double precision

      );

    • id: yellow_truncate_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      TRUNCATE TABLE {{render(vars.staging_table)}};

    • id: yellow_copy_in_to_staging_table
      type: io.kestra.plugin.jdbc.postgresql.CopyIn
      format: CSV
      from: "{{render(vars.data)}}"
      table: "{{render(vars.staging_table)}}"
      header: true
      columns: [VendorID,tpep_pickup_datetime,tpep_dropoff_datetime,passenger_count,trip_distance,RatecodeID,store_and_fwd_flag,PULocationID,DOLocationID,payment_type,fare_amount,extra,mta_tax,tip_amount,tolls_amount,improvement_surcharge,total_amount,congestion_surcharge]

    • id: yellow_add_unique_id_and_filename
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      UPDATE {{render(vars.staging_table)}}
      SET
      unique_row_id = md5(
      COALESCE(CAST(VendorID AS text), '') ||
      COALESCE(CAST(tpep_pickup_datetime AS text), '') ||
      COALESCE(CAST(tpep_dropoff_datetime AS text), '') ||
      COALESCE(PULocationID, '') ||
      COALESCE(DOLocationID, '') ||
      COALESCE(CAST(fare_amount AS text), '') ||
      COALESCE(CAST(trip_distance AS text), '')
      ),
      filename = '{{render(vars.file)}}';

    • id: yellow_merge_data
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      MERGE INTO {{render(vars.table)}} AS T
      USING {{render(vars.staging_table)}} AS S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (
      unique_row_id, filename, VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,
      passenger_count, trip_distance, RatecodeID, store_and_fwd_flag, PULocationID,
      DOLocationID, payment_type, fare_amount, extra, mta_tax, tip_amount, tolls_amount,
      improvement_surcharge, total_amount, congestion_surcharge
      )
      VALUES (
      S.unique_row_id, S.filename, S.VendorID, S.tpep_pickup_datetime, S.tpep_dropoff_datetime,
      S.passenger_count, S.trip_distance, S.RatecodeID, S.store_and_fwd_flag, S.PULocationID,
      S.DOLocationID, S.payment_type, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount,
      S.improvement_surcharge, S.total_amount, S.congestion_surcharge
      );

  • id: if_green_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'green'}}"
    then:

    • id: green_create_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      lpep_pickup_datetime timestamp,
      lpep_dropoff_datetime timestamp,
      store_and_fwd_flag text,
      RatecodeID text,
      PULocationID text,
      DOLocationID text,
      passenger_count integer,
      trip_distance double precision,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      ehail_fee double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      payment_type integer,
      trip_type integer,
      congestion_surcharge double precision

      );

    • id: green_create_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      CREATE TABLE IF NOT EXISTS {{render(vars.staging_table)}} (
      unique_row_id text,
      filename text,
      VendorID text,
      lpep_pickup_datetime timestamp,
      lpep_dropoff_datetime timestamp,
      store_and_fwd_flag text,
      RatecodeID text,
      PULocationID text,
      DOLocationID text,
      passenger_count integer,
      trip_distance double precision,
      fare_amount double precision,
      extra double precision,
      mta_tax double precision,
      tip_amount double precision,
      tolls_amount double precision,
      ehail_fee double precision,
      improvement_surcharge double precision,
      total_amount double precision,
      payment_type integer,
      trip_type integer,
      congestion_surcharge double precision

      );

    • id: green_truncate_staging_table
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      TRUNCATE TABLE {{render(vars.staging_table)}};

    • id: green_copy_in_to_staging_table
      type: io.kestra.plugin.jdbc.postgresql.CopyIn
      format: CSV
      from: "{{render(vars.data)}}"
      table: "{{render(vars.staging_table)}}"
      header: true
      columns: [VendorID,lpep_pickup_datetime,lpep_dropoff_datetime,store_and_fwd_flag,RatecodeID,PULocationID,DOLocationID,passenger_count,trip_distance,fare_amount,extra,mta_tax,tip_amount,tolls_amount,ehail_fee,improvement_surcharge,total_amount,payment_type,trip_type,congestion_surcharge]

    • id: green_add_unique_id_and_filename
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      UPDATE {{render(vars.staging_table)}}
      SET
      unique_row_id = md5(
      COALESCE(CAST(VendorID AS text), '') ||
      COALESCE(CAST(lpep_pickup_datetime AS text), '') ||
      COALESCE(CAST(lpep_dropoff_datetime AS text), '') ||
      COALESCE(PULocationID, '') ||
      COALESCE(DOLocationID, '') ||
      COALESCE(CAST(fare_amount AS text), '') ||
      COALESCE(CAST(trip_distance AS text), '')
      ),
      filename = '{{render(vars.file)}}';

    • id: green_merge_data
      type: io.kestra.plugin.jdbc.postgresql.Queries
      sql: |
      MERGE INTO {{render(vars.table)}} AS T
      USING {{render(vars.staging_table)}} AS S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (
      unique_row_id, filename, VendorID, lpep_pickup_datetime, lpep_dropoff_datetime,
      store_and_fwd_flag, RatecodeID, PULocationID, DOLocationID, passenger_count,
      trip_distance, fare_amount, extra, mta_tax, tip_amount, tolls_amount, ehail_fee,
      improvement_surcharge, total_amount, payment_type, trip_type, congestion_surcharge
      )
      VALUES (
      S.unique_row_id, S.filename, S.VendorID, S.lpep_pickup_datetime, S.lpep_dropoff_datetime,
      S.store_and_fwd_flag, S.RatecodeID, S.PULocationID, S.DOLocationID, S.passenger_count,
      S.trip_distance, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount, S.ehail_fee,
      S.improvement_surcharge, S.total_amount, S.payment_type, S.trip_type, S.congestion_surcharge
      );

  • id: purge_files
    type: io.kestra.plugin.core.storage.PurgeCurrentExecutionFiles
    pluginDefaults:

  • type: io.kestra.plugin.jdbc.postgresql
    values:
    url: jdbc:postgresql://pgdatabase:5432/ny_taxi
    username: root
    password: root

triggers:

  • id: green_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 1 * *"
    inputs:
    taxi: green

  • id: yellow_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 10 1 * *"
    inputs:
    taxi: yellow

File: 02-workflow-orchestration/flows/06_gcp_kv.yaml

id: 06_gcp_kv
namespace: zoomcamp

tasks:

  • id: gcp_project_id
    type: io.kestra.plugin.core.kv.Set
    key: GCP_PROJECT_ID
    kvType: STRING
    value: kestra-sandbox # TODO replace with your project id

  • id: gcp_location
    type: io.kestra.plugin.core.kv.Set
    key: GCP_LOCATION
    kvType: STRING
    value: europe-west2

  • id: gcp_bucket_name
    type: io.kestra.plugin.core.kv.Set
    key: GCP_BUCKET_NAME
    kvType: STRING
    value: your-name-kestra # TODO make sure it's globally unique!

  • id: gcp_dataset
    type: io.kestra.plugin.core.kv.Set
    key: GCP_DATASET
    kvType: STRING
    value: zoomcamp

File: 02-workflow-orchestration/flows/07_gcp_setup.yaml

id: 07_gcp_setup
namespace: zoomcamp

tasks:

  • id: create_gcs_bucket
    type: io.kestra.plugin.gcp.gcs.CreateBucket
    ifExists: SKIP
    storageClass: REGIONAL
    name: "{{kv('GCP_BUCKET_NAME')}}" # make sure it's globally unique!

  • id: create_bq_dataset
    type: io.kestra.plugin.gcp.bigquery.CreateDataset
    name: "{{kv('GCP_DATASET')}}"
    ifExists: SKIP

pluginDefaults:

  • type: io.kestra.plugin.gcp
    values:
    serviceAccount: "{{secret('GCP_CREDS')}}"
    projectId: "{{kv('GCP_PROJECT_ID')}}"
    location: "{{kv('GCP_LOCATION')}}"
    bucket: "{{kv('GCP_BUCKET_NAME')}}"

File: 02-workflow-orchestration/flows/08_gcp_taxi.yaml

id: 08_gcp_taxi
namespace: zoomcamp
The CSV Data used in the course: https://github.com/DataTalksClub/nyc-tlc-data/releases

inputs:

  • id: taxi
    type: SELECT
    displayName: Select taxi type
    values: [yellow, green]
    defaults: green

  • id: year
    type: SELECT
    displayName: Select year
    values: ["2019", "2020"]
    defaults: "2019"
    allowCustomValue: true # allows you to type 2021 from the UI for the homework 🤗

  • id: month
    type: SELECT
    displayName: Select month
    values: ["01", "02", "03", "04", "05", "06", "07", "08", "09", "10", "11", "12"]
    defaults: "01"

variables:
file: "{{inputs.taxi}}tripdata{{inputs.year}}-{{inputs.month}}.csv"
gcs_file: "gs://{{kv('GCP_BUCKET_NAME')}}/{{vars.file}}"
table: "{{kv('GCP_DATASET')}}.{{inputs.taxi}}tripdata{{inputs.year}}_{{inputs.month}}"
data: "{{outputs.extract.outputFiles[inputs.taxi ~ 'tripdata' ~ inputs.year ~ '-' ~ inputs.month ~ '.csv']}}"

tasks:

  • id: set_label
    type: io.kestra.plugin.core.execution.Labels
    labels:
    file: "{{render(vars.file)}}"
    taxi: "{{inputs.taxi}}"

  • id: extract
    type: io.kestra.plugin.scripts.shell.Commands
    outputFiles:

  • id: upload_to_gcs
    type: io.kestra.plugin.gcp.gcs.Upload
    from: "{{render(vars.data)}}"
    to: "{{render(vars.gcs_file)}}"

  • id: if_yellow_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'yellow'}}"
    then:

    • id: bq_yellow_tripdata
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE TABLE IF NOT EXISTS {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.yellow_tripdata
      (
      unique_row_id BYTES OPTIONS (description = 'A unique identifier for the trip, generated by hashing key trip attributes.'),
      filename STRING OPTIONS (description = 'The source filename from which the trip data was loaded.'),
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      tpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      tpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      passenger_count INTEGER OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. TRUE = store and forward trip, FALSE = not a store and forward trip'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      PARTITION BY DATE(tpep_pickup_datetime);

    • id: bq_yellow_table_ext
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE EXTERNAL TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext
      (
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      tpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      tpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      passenger_count INTEGER OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. TRUE = store and forward trip, FALSE = not a store and forward trip'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      OPTIONS (
      format = 'CSV',
      uris = ['{{render(vars.gcs_file)}}'],
      skip_leading_rows = 1,
      ignore_unknown_values = TRUE

      );

    • id: bq_yellow_table_tmp
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}
      AS
      SELECT
      MD5(CONCAT(
      COALESCE(CAST(VendorID AS STRING), ""),
      COALESCE(CAST(tpep_pickup_datetime AS STRING), ""),
      COALESCE(CAST(tpep_dropoff_datetime AS STRING), ""),
      COALESCE(CAST(PULocationID AS STRING), ""),
      COALESCE(CAST(DOLocationID AS STRING), "")
      )) AS unique_row_id,
      "{{render(vars.file)}}" AS filename,
      *

      FROM {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext;

    • id: bq_yellow_merge
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      MERGE INTO {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.yellow_tripdata T
      USING {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}} S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (unique_row_id, filename, VendorID, tpep_pickup_datetime, tpep_dropoff_datetime, passenger_count, trip_distance, RatecodeID, store_and_fwd_flag, PULocationID, DOLocationID, payment_type, fare_amount, extra, mta_tax, tip_amount, tolls_amount, improvement_surcharge, total_amount, congestion_surcharge)
      VALUES (S.unique_row_id, S.filename, S.VendorID, S.tpep_pickup_datetime, S.tpep_dropoff_datetime, S.passenger_count, S.trip_distance, S.RatecodeID, S.store_and_fwd_flag, S.PULocationID, S.DOLocationID, S.payment_type, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount, S.improvement_surcharge, S.total_amount, S.congestion_surcharge);

  • id: if_green_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'green'}}"
    then:

    • id: bq_green_tripdata
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE TABLE IF NOT EXISTS {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.green_tripdata
      (
      unique_row_id BYTES OPTIONS (description = 'A unique identifier for the trip, generated by hashing key trip attributes.'),
      filename STRING OPTIONS (description = 'The source filename from which the trip data was loaded.'),
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      lpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      lpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. Y= store and forward trip N= not a store and forward trip'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      passenger_count INT64 OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      ehail_fee NUMERIC,
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      trip_type STRING OPTIONS (description = 'A code indicating whether the trip was a street-hail or a dispatch that is automatically assigned based on the metered rate in use but can be altered by the driver. 1= Street-hail 2= Dispatch'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      PARTITION BY DATE(lpep_pickup_datetime);

    • id: bq_green_table_ext
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE EXTERNAL TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext
      (
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      lpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      lpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. Y= store and forward trip N= not a store and forward trip'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      passenger_count INT64 OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      ehail_fee NUMERIC,
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      trip_type STRING OPTIONS (description = 'A code indicating whether the trip was a street-hail or a dispatch that is automatically assigned based on the metered rate in use but can be altered by the driver. 1= Street-hail 2= Dispatch'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      OPTIONS (
      format = 'CSV',
      uris = ['{{render(vars.gcs_file)}}'],
      skip_leading_rows = 1,
      ignore_unknown_values = TRUE

      );

    • id: bq_green_table_tmp
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}
      AS
      SELECT
      MD5(CONCAT(
      COALESCE(CAST(VendorID AS STRING), ""),
      COALESCE(CAST(lpep_pickup_datetime AS STRING), ""),
      COALESCE(CAST(lpep_dropoff_datetime AS STRING), ""),
      COALESCE(CAST(PULocationID AS STRING), ""),
      COALESCE(CAST(DOLocationID AS STRING), "")
      )) AS unique_row_id,
      "{{render(vars.file)}}" AS filename,
      *

      FROM {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext;

    • id: bq_green_merge
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      MERGE INTO {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.green_tripdata T
      USING {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}} S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (unique_row_id, filename, VendorID, lpep_pickup_datetime, lpep_dropoff_datetime, store_and_fwd_flag, RatecodeID, PULocationID, DOLocationID, passenger_count, trip_distance, fare_amount, extra, mta_tax, tip_amount, tolls_amount, ehail_fee, improvement_surcharge, total_amount, payment_type, trip_type, congestion_surcharge)
      VALUES (S.unique_row_id, S.filename, S.VendorID, S.lpep_pickup_datetime, S.lpep_dropoff_datetime, S.store_and_fwd_flag, S.RatecodeID, S.PULocationID, S.DOLocationID, S.passenger_count, S.trip_distance, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount, S.ehail_fee, S.improvement_surcharge, S.total_amount, S.payment_type, S.trip_type, S.congestion_surcharge);

  • id: purge_files
    type: io.kestra.plugin.core.storage.PurgeCurrentExecutionFiles
    disabled: false

pluginDefaults:

  • type: io.kestra.plugin.gcp
    values:
    serviceAccount: "{{secret('GCP_CREDS')}}"
    projectId: "{{kv('GCP_PROJECT_ID')}}"
    location: "{{kv('GCP_LOCATION')}}"
    bucket: "{{kv('GCP_BUCKET_NAME')}}"

File: 02-workflow-orchestration/flows/09_gcp_taxi_scheduled.yaml

id: 09_gcp_taxi_scheduled
namespace: zoomcamp
Best to add a label backfill:true from the UI to track executions created via a backfill.
CSV data used here comes from: https://github.com/DataTalksClub/nyc-tlc-data/releases

inputs:

  • id: taxi
    type: SELECT
    displayName: Select taxi type
    values: [yellow, green]
    defaults: green

variables:
file: "{{inputs.taxi}}tripdata{{trigger.date | date('yyyy-MM')}}.csv"
gcs_file: "gs://{{kv('GCP_BUCKET_NAME')}}/{{vars.file}}"
table: "{{kv('GCP_DATASET')}}.{{inputs.taxi}}tripdata{{trigger.date | date('yyyy_MM')}}"
data: "{{outputs.extract.outputFiles[inputs.taxi ~ 'tripdata' ~ (trigger.date | date('yyyy-MM')) ~ '.csv']}}"

tasks:

  • id: set_label
    type: io.kestra.plugin.core.execution.Labels
    labels:
    file: "{{render(vars.file)}}"
    taxi: "{{inputs.taxi}}"

  • id: extract
    type: io.kestra.plugin.scripts.shell.Commands
    outputFiles:

  • id: upload_to_gcs
    type: io.kestra.plugin.gcp.gcs.Upload
    from: "{{render(vars.data)}}"
    to: "{{render(vars.gcs_file)}}"

  • id: if_yellow_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'yellow'}}"
    then:

    • id: bq_yellow_tripdata
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE TABLE IF NOT EXISTS {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.yellow_tripdata
      (
      unique_row_id BYTES OPTIONS (description = 'A unique identifier for the trip, generated by hashing key trip attributes.'),
      filename STRING OPTIONS (description = 'The source filename from which the trip data was loaded.'),
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      tpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      tpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      passenger_count INTEGER OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. TRUE = store and forward trip, FALSE = not a store and forward trip'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      PARTITION BY DATE(tpep_pickup_datetime);

    • id: bq_yellow_table_ext
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE EXTERNAL TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext
      (
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      tpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      tpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      passenger_count INTEGER OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. TRUE = store and forward trip, FALSE = not a store and forward trip'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      OPTIONS (
      format = 'CSV',
      uris = ['{{render(vars.gcs_file)}}'],
      skip_leading_rows = 1,
      ignore_unknown_values = TRUE

      );

    • id: bq_yellow_table_tmp
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}
      AS
      SELECT
      MD5(CONCAT(
      COALESCE(CAST(VendorID AS STRING), ""),
      COALESCE(CAST(tpep_pickup_datetime AS STRING), ""),
      COALESCE(CAST(tpep_dropoff_datetime AS STRING), ""),
      COALESCE(CAST(PULocationID AS STRING), ""),
      COALESCE(CAST(DOLocationID AS STRING), "")
      )) AS unique_row_id,
      "{{render(vars.file)}}" AS filename,
      *

      FROM {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext;

    • id: bq_yellow_merge
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      MERGE INTO {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.yellow_tripdata T
      USING {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}} S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (unique_row_id, filename, VendorID, tpep_pickup_datetime, tpep_dropoff_datetime, passenger_count, trip_distance, RatecodeID, store_and_fwd_flag, PULocationID, DOLocationID, payment_type, fare_amount, extra, mta_tax, tip_amount, tolls_amount, improvement_surcharge, total_amount, congestion_surcharge)
      VALUES (S.unique_row_id, S.filename, S.VendorID, S.tpep_pickup_datetime, S.tpep_dropoff_datetime, S.passenger_count, S.trip_distance, S.RatecodeID, S.store_and_fwd_flag, S.PULocationID, S.DOLocationID, S.payment_type, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount, S.improvement_surcharge, S.total_amount, S.congestion_surcharge);

  • id: if_green_taxi
    type: io.kestra.plugin.core.flow.If
    condition: "{{inputs.taxi == 'green'}}"
    then:

    • id: bq_green_tripdata
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE TABLE IF NOT EXISTS {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.green_tripdata
      (
      unique_row_id BYTES OPTIONS (description = 'A unique identifier for the trip, generated by hashing key trip attributes.'),
      filename STRING OPTIONS (description = 'The source filename from which the trip data was loaded.'),
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      lpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      lpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. Y= store and forward trip N= not a store and forward trip'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      passenger_count INT64 OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      ehail_fee NUMERIC,
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      trip_type STRING OPTIONS (description = 'A code indicating whether the trip was a street-hail or a dispatch that is automatically assigned based on the metered rate in use but can be altered by the driver. 1= Street-hail 2= Dispatch'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      PARTITION BY DATE(lpep_pickup_datetime);

    • id: bq_green_table_ext
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE EXTERNAL TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext
      (
      VendorID STRING OPTIONS (description = 'A code indicating the LPEP provider that provided the record. 1= Creative Mobile Technologies, LLC; 2= VeriFone Inc.'),
      lpep_pickup_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was engaged'),
      lpep_dropoff_datetime TIMESTAMP OPTIONS (description = 'The date and time when the meter was disengaged'),
      store_and_fwd_flag STRING OPTIONS (description = 'This flag indicates whether the trip record was held in vehicle memory before sending to the vendor, aka "store and forward," because the vehicle did not have a connection to the server. Y= store and forward trip N= not a store and forward trip'),
      RatecodeID STRING OPTIONS (description = 'The final rate code in effect at the end of the trip. 1= Standard rate 2=JFK 3=Newark 4=Nassau or Westchester 5=Negotiated fare 6=Group ride'),
      PULocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was engaged'),
      DOLocationID STRING OPTIONS (description = 'TLC Taxi Zone in which the taximeter was disengaged'),
      passenger_count INT64 OPTIONS (description = 'The number of passengers in the vehicle. This is a driver-entered value.'),
      trip_distance NUMERIC OPTIONS (description = 'The elapsed trip distance in miles reported by the taximeter.'),
      fare_amount NUMERIC OPTIONS (description = 'The time-and-distance fare calculated by the meter'),
      extra NUMERIC OPTIONS (description = 'Miscellaneous extras and surcharges. Currently, this only includes the $0.50 and $1 rush hour and overnight charges'),
      mta_tax NUMERIC OPTIONS (description = '$0.50 MTA tax that is automatically triggered based on the metered rate in use'),
      tip_amount NUMERIC OPTIONS (description = 'Tip amount. This field is automatically populated for credit card tips. Cash tips are not included.'),
      tolls_amount NUMERIC OPTIONS (description = 'Total amount of all tolls paid in trip.'),
      ehail_fee NUMERIC,
      improvement_surcharge NUMERIC OPTIONS (description = '$0.30 improvement surcharge assessed on hailed trips at the flag drop. The improvement surcharge began being levied in 2015.'),
      total_amount NUMERIC OPTIONS (description = 'The total amount charged to passengers. Does not include cash tips.'),
      payment_type INTEGER OPTIONS (description = 'A numeric code signifying how the passenger paid for the trip. 1= Credit card 2= Cash 3= No charge 4= Dispute 5= Unknown 6= Voided trip'),
      trip_type STRING OPTIONS (description = 'A code indicating whether the trip was a street-hail or a dispatch that is automatically assigned based on the metered rate in use but can be altered by the driver. 1= Street-hail 2= Dispatch'),
      congestion_surcharge NUMERIC OPTIONS (description = 'Congestion surcharge applied to trips in congested zones')

      )
      OPTIONS (
      format = 'CSV',
      uris = ['{{render(vars.gcs_file)}}'],
      skip_leading_rows = 1,
      ignore_unknown_values = TRUE

      );

    • id: bq_green_table_tmp
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      CREATE OR REPLACE TABLE {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}
      AS
      SELECT
      MD5(CONCAT(
      COALESCE(CAST(VendorID AS STRING), ""),
      COALESCE(CAST(lpep_pickup_datetime AS STRING), ""),
      COALESCE(CAST(lpep_dropoff_datetime AS STRING), ""),
      COALESCE(CAST(PULocationID AS STRING), ""),
      COALESCE(CAST(DOLocationID AS STRING), "")
      )) AS unique_row_id,
      "{{render(vars.file)}}" AS filename,
      *

      FROM {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}_ext;

    • id: bq_green_merge
      type: io.kestra.plugin.gcp.bigquery.Query
      sql: |
      MERGE INTO {{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.green_tripdata T
      USING {{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}} S
      ON T.unique_row_id = S.unique_row_id
      WHEN NOT MATCHED THEN
      INSERT (unique_row_id, filename, VendorID, lpep_pickup_datetime, lpep_dropoff_datetime, store_and_fwd_flag, RatecodeID, PULocationID, DOLocationID, passenger_count, trip_distance, fare_amount, extra, mta_tax, tip_amount, tolls_amount, ehail_fee, improvement_surcharge, total_amount, payment_type, trip_type, congestion_surcharge)
      VALUES (S.unique_row_id, S.filename, S.VendorID, S.lpep_pickup_datetime, S.lpep_dropoff_datetime, S.store_and_fwd_flag, S.RatecodeID, S.PULocationID, S.DOLocationID, S.passenger_count, S.trip_distance, S.fare_amount, S.extra, S.mta_tax, S.tip_amount, S.tolls_amount, S.ehail_fee, S.improvement_surcharge, S.total_amount, S.payment_type, S.trip_type, S.congestion_surcharge);

  • id: purge_files
    type: io.kestra.plugin.core.storage.PurgeCurrentExecutionFiles
    pluginDefaults:

  • type: io.kestra.plugin.gcp
    values:
    serviceAccount: "{{secret('GCP_CREDS')}}"
    projectId: "{{kv('GCP_PROJECT_ID')}}"
    location: "{{kv('GCP_LOCATION')}}"
    bucket: "{{kv('GCP_BUCKET_NAME')}}"

triggers:

  • id: green_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 1 * *"
    inputs:
    taxi: green

  • id: yellow_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 10 1 * *"
    inputs:
    taxi: yellow

File: 02-workflow-orchestration/flows/10_chat_without_rag.yaml

id: 10_chat_without_rag
namespace: zoomcamp
This flow demonstrates what happens when you query an LLM WITHOUT RAG.
The model can only rely on its training data, which may be outdated or incomplete.

After running this, check out 11_chat_with_rag.yaml to see how RAG fixes these issues.

tasks:

  • id: chat_without_rag
    type: io.kestra.plugin.ai.completion.ChatCompletion
    provider:
    type: io.kestra.plugin.ai.provider.GoogleGemini
    modelName: gemini-2.5-flash
    apiKey: "{{ kv('GEMINI_API_KEY') }}"
    messages:

    • type: USER
      content: |
      Which features were released in Kestra 1.1?
      Please list at least 5 major features with brief descriptions.

  • id: log_results
    type: io.kestra.plugin.core.log.Log
    message: |
    ❌ Response WITHOUT RAG (no retrieved context):
    {{ outputs.chat_without_rag.textOutput }}

    🤔 Did you notice that this response seems to be:

    • Incorrect
    • Vague/generic
    • Listing features that haven't been added in exactly this version but rather a long time ago

    👉 This is why context matters. Run 11_chat_with_rag.yaml to see the accurate, context-grounded response.

File: 02-workflow-orchestration/flows/11_chat_with_rag.yaml

id: 11_chat_with_rag
namespace: zoomcamp
This flow demonstrates RAG (Retrieval Augmented Generation) by ingesting Kestra release documentation and using it to answer questions accurately.

Compare this with 10_chat_without_rag.yaml to see the difference RAG makes.

tasks:

  • id: ingest_release_notes
    type: io.kestra.plugin.ai.rag.IngestDocument
    provider:
    type: io.kestra.plugin.ai.provider.GoogleGemini
    modelName: gemini-embedding-001
    apiKey: "{{ kv('GEMINI_API_KEY') }}"
    embeddings:
    type: io.kestra.plugin.ai.embeddings.KestraKVStore
    drop: true
    fromExternalURLs:

  • id: chat_with_rag
    type: io.kestra.plugin.ai.rag.ChatCompletion
    chatProvider:
    type: io.kestra.plugin.ai.provider.GoogleGemini
    modelName: gemini-2.5-flash
    apiKey: "{{ kv('GEMINI_API_KEY') }}"
    embeddingProvider:
    type: io.kestra.plugin.ai.provider.GoogleGemini
    modelName: gemini-embedding-001
    apiKey: "{{ kv('GEMINI_API_KEY') }}"
    embeddings:
    type: io.kestra.plugin.ai.embeddings.KestraKVStore
    systemMessage: |
    You are a helpful assistant that answers questions about Kestra.
    Use the provided documentation to give accurate, specific answers.
    If you don't find the information in the context, say so.
    prompt: |
    Which features were released in Kestra 1.1?
    Please list at least 5 major features with brief descriptions.

  • id: log_results
    type: io.kestra.plugin.core.log.Log
    message: |
    ✅ RAG Response (with retrieved context):
    {{ outputs.chat_with_rag.textOutput }}

    Note that this response is detailed, accurate, and grounded in the actual release documentation. Compare this with the output from 06_chat_without_rag.yaml.

File: 03-data-warehouse/README.md

Data Warehousing

Units

  1. Data Warehouse and BigQuery
  2. Partitioning vs Clustering
  3. BigQuery Best Practices
  4. Internals of BigQuery
  5. Machine Learning in BigQuery
  6. Deploying a Machine Learning Model from BigQuery

Homework

Other files in this module

  • extras/ — a standalone script that downloads the taxi
    CSVs and uploads them to Cloud Storage as parquet, without an orchestrator

Community notes

Did you take notes? You can share them here

File: 03-data-warehouse/01-data-warehouse-and-bigquery.md

video_url: https://www.youtube.com/watch?v=jrHljAoD6nM prev_url: ../02-workflow-orchestration/18-additional-resources.md next_url: 02-partitioning-vs-clustering.md

Data Warehouse and BigQuery

In this unit we talk about data warehouses in general and BigQuery as
our example: how analytical databases differ from the transactional
ones behind everyday apps, what a data warehouse looks like
architecturally, and a first hands-on tour of BigQuery — the interface,
public datasets, external tables, pricing, partitioning and clustering.

OLTP vs OLAP

Databases come in two flavors, and the difference starts with the use
case.

OLTP stands for online transaction processing. This is the kind of
database you use in backend services, where you group a couple of SQL
queries together into one transaction and roll back in case one of them
fails. A classic example is a shop: the order, the payment and the
inventory update either all happen or none of them do.

OLAP stands for online analytical processing. These databases are
designed for a different purpose: putting a lot of data in and
discovering hidden insights. They are mainly used for analytical work
by data analysts and data scientists.

The two types differ in almost every aspect:

  • In OLTP the updates are fast but small. In OLAP the data is
    periodically refreshed, and the data size is generally way larger.
  • OLTP databases are normalized for efficiency. OLAP databases are
    denormalized for analysis.
  • OLTP increases productivity for the end user. OLAP increases
    productivity for data analysts and executives.
  • OLTP users are customer-facing personnel, clerks and online shoppers.
    OLAP users are knowledge workers: data analysts, business analysts
    and executives.
OLTP OLAP
Purpose Control and run essential business operations in real time Plan, solve problems, support decisions, discover hidden insights
Data updates Short, fast updates initiated by user Data periodically refreshed with scheduled, long-running batch jobs
Database design Normalized databases for efficiency Denormalized databases for analysis
Space requirements Generally small if historical data is archived Generally large due to aggregating large datasets

What is a data warehouse

A data warehouse is an OLAP solution used for reporting and data
analysis.

A warehouse generally consists of raw data, metadata and summary data.
It has many data sources — operational systems, flat files, maybe an
OLTP database — and all these systems report to a staging area, which
is then written to the data warehouse.

As an output, a data warehouse can be transformed into data marts:
smaller, subject-specific slices such as purchasing, sales and
inventory, which different users can access directly. For analysts,
using data marts as an interface is the ideal situation. But in certain
use cases, especially for data scientists, it makes sense to look at
the raw data directly from the warehouse. A data warehouse provides all
these possibilities.

BigQuery

BigQuery is a data warehouse solution, and its biggest advantage is
that it is serverless: there are no servers to manage and no database
software to install.

This matters because when a company starts its data warehouse journey,
a huge chunk of time normally goes into creating and maintaining the
warehouse. BigQuery removes that pain: it provides the software as well
as the infrastructure, with scalability and high availability in mind.
You can start with a couple of gigabytes of data and scale easily to
petabytes without issues.

Some of the built-in features that really outshine:

  • machine learning via the SQL interface of BigQuery
  • handling of geospatial data
  • business intelligence queries

Another big advantage is the flexibility of how it stores data. In a
classic setup one big server holds storage and compute together, so
once your data size increases, the machine has to grow with it.
BigQuery separates the compute engine from storage and analyzes your
data on a storage of your choice. That is a huge win in terms of cost.

How much BigQuery costs

BigQuery has two pricing models.

On-demand pricing is based on the amount of data you scan or process:
every terabyte processed costs around $5.

Flat rate pricing is based on the number of pre-requested slots —
BigQuery's units of processing capacity. 100 slots cost roughly
$2,000 per month, which converts to about 400 TB of data processed
under on-demand pricing. So flat rate does not make sense unless you
are scanning way above 200 TB of data.

Slots also change how the system behaves under load. If you have 50
queries running and all your 100 slots are full, the 51st query has to
wait. That would not happen on demand: there BigQuery gives you more
slots based on the requirements of a query.

Public datasets

Before working with BigQuery, two things worth knowing. First, BigQuery
generally caches query results — for the demos in this module caching
was disabled to get consistent results. Second, BigQuery ships a lot of
open source public data you can query out of the box.

One example is the New York City Citi Bike stations data. You can
search any dataset by table name in the search bar, and open the table
to see its columns: station id, name, short name, latitude, longitude
and others. This table has only 1,584 rows and its size is just a few
KB — a very small dataset. You can also do a quick preview to
see how the data is arranged.

Let's query it:

sql
SELECT
    station_id,
    name
FROM bigquery-public-data.new_york_citibike.citibike_stations
LIMIT 100;

The station ids and names show up in the query results panel at the
bottom. From there you can save the results as CSV, or explore them
further in Data Studio.

External tables

For this module we use the NYC taxi trip data, which was already
uploaded to Google Cloud Storage. BigQuery lets you create an external
table over such files: the data itself stays in Cloud Storage, and
BigQuery keeps only the metadata.

The interface organizes everything into projects, datasets and tables:
taxi-rides-ny is the project, nytaxi is the dataset, and
external_yellow_tripdata is a table in it. The bucket holds trip data
CSVs for 2019, 2020 and so on, and the external table points at the
2019 and 2020 files:

sql
CREATE OR REPLACE EXTERNAL TABLE `taxi-rides-ny.nytaxi.external_yellow_tripdata`
OPTIONS (
    format = 'CSV',
    uris = [
        'gs://nyc-tl-data/trip data/yellow_tripdata_2019-*.csv',
        'gs://nyc-tl-data/trip data/yellow_tripdata_2020-*.csv'
    ]
);

Once the table is created, open it: BigQuery has read the column names
of the CSVs and already understands the types, and it figures out which
columns are nullable. You do not have to define the schema — though you
can if you want to.

Now look at the details of the table: long-term storage is 0 bytes, the
table size is 0 bytes, and the number of rows is 0. When you create an
external table, BigQuery cannot determine its rows or size — the data
is not inside BigQuery, it is in an external system, Google Cloud
Storage.

The resulting external table has the following details:

Field Value
Table ID taxi-rides-ny.nytaxi.external_yellow_tripdata
Table size 0 B
Long-term storage size 0 B
Number of rows 0
Table expiration Never
Data location europe-west3
Source URI gs://nyc-tlc/data/trip_data/yellow_tripdata_2019-*.csv
Source URI gs://nyc-tlc/data/trip_data/yellow_tripdata_2020-*.csv
Auto-detect schema true
Source format CSV

Querying it works like any other table:

sql
SELECT *
FROM taxi-rides-ny.nytaxi.external_yellow_tripdata
LIMIT 10;

You get the vendor id, pickup and dropoff datetimes, passenger count,
trip distance, location information, and the fare and total amounts.

Partitioning

Partitioning is a very useful BigQuery feature. Take a dataset of Stack
Overflow questions from 2018 with columns like creation date, title and
tags. Suppose most of your queries filter on date — give me the count
of questions asked only in March, or only in the first week of March.
Partitioning by the creation date can really improve performance.

When you partition the table by creation date, each date becomes its
own partition: all rows created on 1 March 2018 go into the first
partition, 2 March into the second, and so on. This is powerful because
once BigQuery understands it only needs the data for 2 March 2018, it
will not read or process the data for 1 March or 3 March. Processing
less data reduces cost.

Back to the taxi data. Until now we only had the external table, which
is not partitioned. First let's create a plain, non-partitioned table
by copying the external table's content:

sql
CREATE OR REPLACE TABLE taxi-rides-ny.nytaxi.yellow_tripdata_non_partitioned AS
SELECT *
FROM taxi-rides-ny.nytaxi.external_yellow_tripdata;

This takes some time, because the data is copied from Google Cloud
Storage into actual BigQuery storage. The partitioned version adds only
one line, PARTITION BY:

sql
CREATE OR REPLACE TABLE taxi-rides-ny.nytaxi.yellow_tripdata_partitioned
PARTITION BY DATE(tpep_pickup_datetime) AS
SELECT *
FROM taxi-rides-ny.nytaxi.external_yellow_tripdata;

The partitioned table knows its size — around 13-14 GB — and the
details show it is partitioned by day on tpep_pickup_datetime. In the
schema view you can also spot the difference: a non-partitioned table
is one solid block of columns, while a partitioned table has a break in
between. That is a quick way to tell which tables are partitioned.

Now the payoff. Run the same query against both tables — the distinct
vendor ids for June 2019:

sql
SELECT DISTINCT(VendorID)
FROM taxi-rides-ny.nytaxi.yellow_tripdata_non_partitioned
WHERE DATE(tpep_pickup_datetime) BETWEEN '2019-06-01' AND '2019-06-30';

On the non-partitioned table, BigQuery estimates it will process 1.6 GB
— almost all the data in the table. Change only the table name to the
partitioned one, and the estimate drops to about 106 MB. If you run
this query over and over, you process 106 MB each time instead of
1.6 GB, which directly reduces your cost.

The partitioned query's estimate and actual processed bytes are:

Query Bytes processed
Estimate before running 105.9 MiB
Actual after running 105.9 MiB

You can also inspect the partitions themselves. Every dataset has an
INFORMATION_SCHEMA with a PARTITIONS view:

sql
SELECT
    table_name,
    partition_id,
    total_rows
FROM `nytaxi.INFORMATION_SCHEMA.PARTITIONS`
WHERE table_name = 'yellow_tripdata_partitioned'
ORDER BY total_rows DESC;

This shows how many rows fall into which partition: in our table,
1 February 2019 has the most rows, and 5 April 2019 has around 292,000
rows. It is also a quick check for bias — whether some partitions are
getting much more data than others.

Clustering

From partitioning we move to another interesting concept: clustering.

Back to the Stack Overflow example. We partitioned the table by date.
Now we also cluster it by tag. Within each partition, rows with the
same tag are stored together: in the second partition, the android rows
come first, then the linux rows; in the first partition, android is
grouped together, linux comes after, and sql after that.

Because related rows sit next to each other, BigQuery can skip data
inside a partition too. This improves cost as well as query
performance.

For the taxi data, we create a table that is both partitioned and
clustered:

sql
CREATE OR REPLACE TABLE taxi-rides-ny.nytaxi.yellow_tripdata_partitioned_clustered
PARTITION BY DATE(tpep_pickup_datetime)
CLUSTER BY VendorID AS
SELECT * 
FROM taxi-rides-ny.nytaxi.external_yellow_tripdata;

Why VendorID? Because of how the data will be queried: in this use
case the data is always filtered on vendor id and pickup date, and this
partitioning and clustering targets exactly those filters. The details
view confirms it: partitioned by day on tpep_pickup_datetime, and
clustered by VendorID.

Now compare the two tables by counting all trips from vendor 1 between
1 June 2019 and 31 December 2020:

sql
SELECT count(*) as trips
FROM taxi-rides-ny.nytaxi.yellow_tripdata_partitioned
WHERE DATE(tpep_pickup_datetime) BETWEEN '2019-06-01' AND '2020-12-31'
  AND VendorID=1;

Before running, BigQuery estimates 1.1 GB. After running: 0.7 seconds,
and 1.1 GB of data actually processed. Here is a rule to remember:
whatever BigQuery shows before the run is an approximation — the actual
processed bytes only appear after the run.

Run the same query on the partitioned and clustered table. The estimate
still says 1.1 GB — the approximation cannot know what clustering will
skip. But the actual run processes less: 843.5 MB instead of 1.1 GB.
That is the clustering effect.

The comparison between the partitioned and clustered queries is:

Table Estimated bytes Actual bytes
Partitioned 1.1 GB 1.1 GB
Partitioned and clustered 1.1 GB 843.5 MB

When to prefer partitioning, when clustering, and when both — that is
the topic of the next unit.

Materials

  • Slides
    for the whole module
  • big_query.sql — the SQL used through the module. The first
    statements belong to this unit: querying a BigQuery public dataset, and
    creating an external table over the taxi CSVs in Cloud Storage.

File: 03-data-warehouse/02-partitioning-vs-clustering.md

video_url: https://www.youtube.com/watch?v=-CqXf7vhhDs prev_url: 01-data-warehouse-and-bigquery.md next_url: 03-bigquery-best-practices.md

Partitioning vs Clustering

In the previous unit we created
partitioned and clustered versions of the yellow taxi data table and saw
how much less data our queries scanned as a result. This unit goes into
the detail: the partitioning options BigQuery gives you, how clustering
sorts your data, and how to decide which of the two to use.

Partitioning options

When you create a partitioned table in BigQuery, you choose what the
table is partitioned by. There are three options:

  • Time-unit column: a column with a timestamp or date, like the pickup
    time of a taxi ride.
  • Ingestion time: BigQuery assigns each row to a partition based on
    when it was inserted, using a pseudo-column called _PARTITIONTIME.
  • Integer range: an integer column split into ranges.

With a time-unit column or ingestion time you also pick the
granularity: daily (the default), hourly, or monthly and yearly.

Daily is a good way to start. It works well when the data is of medium
size and spread evenly across the days.

Hourly is for a huge amount of data coming in, when you want to process
it by the hour. Keep an eye on the number of partitions this creates:
BigQuery allows at most 4000 partitions per table. With hourly
partitions you might need an expiration strategy that removes old
partitions.

Monthly or yearly is for the opposite case: a small amount of data
spread across a wide range of dates.

How clustering works

When you cluster a table, the columns you specify are used to colocate
related data - rows with similar values end up stored together.

The order of the columns is important, because it determines the sort
order of the data. If you cluster on columns a, b and c, the table is
sorted first by a, then by b, then by c.

Clustering improves filter and aggregate queries, especially when you
filter or aggregate on the columns you clustered by.

Partitioning and clustering are not free, though. If your table is
small - less than 1 GB - neither shows a significant improvement in
query performance. They actually add cost: partitioned and clustered
tables incur metadata reads and metadata maintenance. For a small table
it can make more sense to have no partitioning or clustering at all.

You can specify up to four clustering columns. They must be top-level,
non-repeated columns, and you can choose from these types: DATE,
BOOLEAN, GEOGRAPHY, INT64, NUMERIC, STRING and DATETIME.

Choosing between partitioning and clustering

Sometimes you want one, sometimes the other, sometimes both. The choice
comes down to a few criteria.

Cost. With partitioning, the cost benefit is known upfront: a query
that filters on the partition column only reads some of the partitions.
With clustering, the benefit is unknown until you run the query. This
matters because BigQuery lets you specify a maximum cost for a query:
if it would exceed that amount, BigQuery does not execute it at all.
You can only do that when the cost is known upfront, which means
partitioning.

Granularity. Use clustering when you need more granularity than
partitioning alone can provide.

Management. Partitioning gives you partition-level management: you can
delete partitions or move partitions between storages. Clustering
gives you none of that.

Columns. With clustering, you commonly filter or aggregate your data
on multiple columns. Partitioning works on a single column only,
because partitioning is only possible on one column.

Cardinality. Use clustering when the number of distinct values in a
column or group of columns is large. That cardinality is a hindrance
for partitioning, because of the 4000-partition limit.

When clustering beats partitioning

Putting the criteria together, these are the situations where you
should choose clustering over partitioning:

  • Partitioning would leave a small amount of data per partition -
    roughly less than 1 GB.
  • Partitioning would create more partitions than the limit of 4000 per
    table.
  • Your mutation operations would modify the majority of the partitions
    frequently - for example, writing to the table every few minutes.

Automatic reclustering

As data is added to a clustered table, the newly inserted rows are
written into blocks whose key ranges can overlap the key ranges stored
in previous blocks. These overlapping keys weaken the sort property of
the table, and queries get slower.

To maintain the performance characteristics of a clustered table,
BigQuery performs automatic reclustering in the background: it restores
the sort property of the table without you doing anything. For
partitioned tables, clustering is maintained within the scope of each
partition.

Reclustering does not impact query performance, and it does not cost
you anything - BigQuery takes care of it.

Summary

In this unit we have seen the partitioning options BigQuery offers, how
clustering colocates and sorts the data, what the criteria are for
choosing between partitioning and clustering, and how BigQuery
reclusters automatically.

A good exercise: think about the yellow taxi dataset from the previous
unit. Which column would you partition it by, and what would you
cluster it on? Think about the queries you want to run - that is what
the choice comes down to.

Materials

  • big_query.sql — the partitioning and clustering half of the
    file: creating partitioned and partitioned-plus-clustered tables from the
    external table, inspecting INFORMATION_SCHEMA.PARTITIONS, and the paired
    queries that show how much data each variant scans.

File: 03-data-warehouse/03-bigquery-best-practices.md

video_url: https://www.youtube.com/watch?v=k81mLJVX08w prev_url: 02-partitioning-vs-clustering.md next_url: 04-internals-of-bigquery.md

BigQuery Best Practices

This unit is a short checklist of best practices for running queries in
BigQuery. Everything on it serves one of two goals: reducing cost or
improving query performance.

Cost reduction

Most of our effort goes into cost reduction or query performance, and the
two are related: the cheaper query is usually the one that reads less data.

The slide lists five habits:

  • Avoid SELECT *. BigQuery stores data columnar: each column is stored
    separately, not as part of a row. When you name the columns you need,
    BigQuery reads only those columns. With * it has to read all of them,
    even the ones you never look at. For one or two columns, this is the
    difference between reading almost nothing and reading the whole table.
  • Price your queries before running them. The estimated price shows up in
    the top right corner of the query editor. Look at it before you press
    run.
  • Use clustered or partitioned tables. We covered this in detail in the
    previous unit: both let BigQuery skip
    large parts of the table.
  • Use streaming inserts with caution. Streaming inserts can increase your
    cost drastically.
  • Materialize query results in stages. Suppose you write a query with a
    CTE (a common table expression, the WITH clause) and reuse it in
    several places. Instead of recomputing it each time, run it once, store
    the result in a table, and use that table in the following steps.

One more thing: BigQuery caches query results. When you rerun the same
query, it can serve the cached result instead of recomputing it.

Query performance

The second group of best practices is about making queries fast.

  • Always filter on the partitioned column, or on the clustered column.
    Otherwise the partitioning and clustering you set up do you no good.
  • Denormalize your data. In a normalized database you split data into many
    small tables to avoid duplication; in a warehouse you often do the
    opposite and keep things together, so a query doesn't have to join them
    back.
  • If you have a complicated structure, use nested or repeated columns.
    They let you keep related data together in one table without
    denormalizing it away entirely.
  • Use external data sources appropriately, but don't overdo it. Reading
    from Google Cloud Storage, for example, can incur more cost than reading
    from BigQuery's own storage.
  • Reduce the data before you use it in a JOIN: filter first, join after.
  • Do not treat WITH clauses as prepared statements. A WITH clause names
    a subquery inside one query, it is not something you define once and
    reuse across queries.
  • Avoid oversharding tables: sharding is splitting your data into many
    small tables, for example one per day. Too many small tables work worse
    than one partitioned table.

More performance tips

A few more things that speed up queries:

  • Avoid JavaScript user-defined functions (UDFs, functions you write
    yourself in SQL).
  • Use approximate aggregation functions rather than the exact ones, such
    as HyperLogLog++ for counting distinct values.
  • ORDER statements should be the last part of your query.
  • Optimize your join patterns. Place the table with the largest number of
    rows first, then the table with the fewest rows, then the remaining
    tables by decreasing size. The largest table gets distributed evenly
    across the nodes, and the next table gets broadcast to all of them.

The reason the join order matters is not obvious until you see how
BigQuery actually executes a query. We look at that in detail in the
next unit.

File: 03-data-warehouse/04-internals-of-bigquery.md

video_url: https://www.youtube.com/watch?v=eduHi1inM4s prev_url: 03-bigquery-best-practices.md next_url: 05-machine-learning-in-bigquery.md

Internals of BigQuery

In this unit we look inside BigQuery: how it stores data, how storage and
compute talk to each other, and how a query gets executed. You don't need
the internals to use BigQuery — the best practices
and partitioning and clustering are enough for day-to-day work. But
understanding how a data warehouse like this is built will help you when
you design data products of your own.

The diagram above is the high-level architecture. Three names to remember:

  • Colossus — the storage
  • Jupiter — the network between storage and compute
  • Dremel — the query execution engine

Colossus: storage and compute are separate

BigQuery stores your data in a separate storage system called Colossus.
Colossus is cheap storage that keeps data in a columnar format — each
column stored on its own, which we'll see more of below.

This separation of storage from compute is the big design decision, and it
significantly reduces cost. If your data size grows tomorrow, you only pay
for storing it in Colossus, which is very cheap. The expensive part is
reading the data and running queries — that's compute, and it only happens
when you actually run something.

Jupiter: the network in between

The separation raises an obvious question. Compute and storage sit on
different hardware — how do they communicate? If the network between them
were slow, every query would spend its time waiting for data, and query
times would suffer.

That's what the Jupiter network solves. Jupiter is the network inside
BigQuery data centers, and it provides approximately one terabyte per
second. With that much bandwidth, compute and storage can live on separate
hardware and still talk to each other without any delays.

Column-oriented storage

Before we get to how queries execute, a quick look at how the data itself
is laid out. The video calls this part Polymer — the component on the
storage side — and compares record-oriented storage with column-oriented
storage.

On the left is record-oriented storage: each record (row r1, r2) is
stored as one piece. This is very similar to structures like CSV, and it's
easy to process and understand.

On the right is column-oriented storage. There, a row appears in multiple
places: each column is stored separately from the others.

BigQuery uses the column-oriented structure, and it gives a huge gain.
Column-oriented storage makes aggregations over columns much faster, and
in a warehouse we rarely query all the columns at once anyway. The typical
requirement is to read a few columns and filter and aggregate on others —
with columns stored separately, the rest of the table is never touched.
That's also why SELECT * is so expensive.

Dremel: executing a query as a tree

Dremel is the query execution engine. It takes your query and divides it
into a tree structure, in such a way that each node executes an individual
subset of the query.

Let's walk through the example in the diagram. Assume a query like
SELECT A, COUNT(B) FROM T GROUP BY A — count the rows per group of
column A.

  • The root server receives the query. It understands what the query wants
    and knows how to divide it into smaller sub-modules. In our case,
    SELECT A, COUNT(B) becomes SELECT A, SUM(C): instead of counting,
    we'll sum up partial counts that the lower levels produce. The rewritten
    query is divided further and sent down as slices R1 to R1+n.
  • The mixers (the intermediate level) receive the modified query, divide
    it further into subsets, and pass those to the leaf nodes.
  • The leaf nodes are the ones that actually talk to the Colossus database.
    They fetch the data, execute the appropriate operations on their slice,
    and return the result to the mixers.
  • The mixers return their results to the root server, where everything is
    aggregated and returned to you as the final result.

This distribution of workers is the reason BigQuery is so fast. If all the
work had to be done on a single node, query time would grow very badly as
data sizes increase. Because BigQuery divides the query into smaller
chunks and pushes them down to the leaf nodes, it scales.

References

The video ends with a reference slide, and the links there are worth
reading — some cover Dremel internals, others BigQuery internals in
general:

File: 03-data-warehouse/05-machine-learning-in-bigquery.md

video_url: https://www.youtube.com/watch?v=B-WtpB0PuG4 prev_url: 04-internals-of-bigquery.md next_url: 06-deploying-a-machine-learning-model.md

Machine Learning in BigQuery

In this unit we train, evaluate and explain a machine learning model without
leaving BigQuery. We build a linear regression model that predicts taxi tip
amounts - all in SQL - and finish with hyperparameter tuning. In the
next unit we export this model
and run it inside Docker.

The queries we run are collected in big_query_ml.sql,
so you don't need to copy them from the video.

Why machine learning in BigQuery

BigQuery ML targets data analysts and managers. You don't have to know Python
or Java: SQL plus some knowledge of machine learning algorithms is enough.

The other advantage is that the data stays in the warehouse. Normally you
would export the data from the data warehouse, build and train a model in a
separate system, and then deploy it. BigQuery lets us build the model inside
the warehouse itself, so the extra export step disappears.

One warning before we start: this and the next video are advanced steps for
BigQuery. If you are not familiar with machine learning, feel free to skip
them.

Pricing

Pricing is a real condition when you choose between using machine learning in
BigQuery and building a model separately. At the time of recording:

  • Gigabytes of data storage are free.
  • One terabyte of queries processed per month is free.
  • When using the CREATE MODEL step, the first 10 gigabytes per month are
    also free.

After the free tier you pay around 250 dollars per terabyte for creating
logistic regression, linear regression, clustering and time series models,
and around 5 dollars per terabyte for AutoML, DNN and boosted tree models,
with extra Vertex AI training costs on top. These are US prices - if you are
in a different region, check the costs for your region.

The machine learning workflow

Before creating a model, it helps to recall the steps of machine learning
development. First we collect data - that part is done, we loaded the taxi
data into BigQuery in the previous units. Then we process the data and do
feature engineering, split it into training and test sets, build a model
with a suitable algorithm, optimize its parameters (hyperparameter tuning),
and validate it against error metrics. Once we are satisfied with the model,
we deploy it.

BigQuery helps with every step. It does feature engineering, including
automatic feature preprocessing. It splits the data into training and
evaluation sets by itself. It offers a choice of algorithms, hyperparameter
tuning, and many error metrics to validate against. And it lets us deploy
the trained model using a Docker image - that is the next unit.

Choosing an algorithm

Which algorithm fits which use case? The BigQuery documentation answers this
with a decision diagram:

Predicting a number - sales figures, stock prices - calls for linear
regression, boosted trees, AutoML, a DNN regressor or wide-and-deep
regression. Customer segmentation is a job for k-means. Predicting a
category - identifying spam, for example - calls for logistic regression,
boosted tree classifiers or AutoML tables. You can always come back to this
diagram, find the use case you are trying to solve, and pick the algorithms
suggested for it.

In this video we build a linear regression model: we predict the tip amount
of a taxi ride, which is a number.

Picking the features

Our dataset is yellow_tripdata_partitioned from the previous units. The
label - the thing we want to predict - is tip_amount, the tip paid to the
driver at the end of the journey.

For the features, we select a handful of columns from the ride:

sql
SELECT passenger_count, trip_distance, PULocationID, DOLocationID, payment_type, fare_amount, tolls_amount, tip_amount
FROM `taxi-rides-ny.nytaxi.yellow_tripdata_partitioned` WHERE fare_amount != 0;

The WHERE fare_amount != 0 filter is there because a lot of rides have a
fare amount of zero, and those rides almost always have a tip amount of zero
too - they would teach the model the wrong thing.

Feature preprocessing

BigQuery ML does feature preprocessing in two ways: automatic and manual.

Automatic preprocessing applies standard transformations for us:
standardization of numeric fields, one-hot encoding of categorical fields,
and multi-hot encoding of arrays. One-hot encoding turns a category feature
into a sparse vector - one position per category, with a 1 in the position
of the value at hand.

Manual preprocessing covers bucketization, polynomial expansion, feature
crossing, n-grams, min-max scaling and more. Our example needs no manual
preprocessing - the automatic one is enough.

But there is a catch. PULocationID, DOLocationID and payment_type are
stored as integers, yet they are not really numbers - they are categories.
payment_type stands for a payment method like card or cash, and location
ID 264 does not mean the location is "264 times" anything. Left as integers,
BigQuery would standardize them like numbers instead of one-hot encoding
them.

Preparing the feature table

So we create a separate table with the same columns, but cast the three
category columns to STRING. A string column is treated as a categorical
feature, and BigQuery will one-hot encode it automatically:

sql
CREATE OR REPLACE TABLE `taxi-rides-ny.nytaxi.yellow_tripdata_ml` (
`passenger_count` INTEGER,
`trip_distance` FLOAT64,
`PULocationID` STRING,
`DOLocationID` STRING,
`payment_type` STRING,
`fare_amount` FLOAT64,
`tolls_amount` FLOAT64,
`tip_amount` FLOAT64
) AS (
SELECT passenger_count, trip_distance, cast(PULocationID AS STRING), CAST(DOLocationID AS STRING),
CAST(payment_type AS STRING), fare_amount, tolls_amount, tip_amount
FROM `taxi-rides-ny.nytaxi.yellow_tripdata_partitioned` WHERE fare_amount != 0
);

Running this processed about 6.5 GB of data and produced a table with the
correct types for our machine learning model:

Creating the model

Now the actual model. We name it tip_model and use linear regression as
the algorithm. The label column is tip_amount, and the data split method
'AUTO_SPLIT' tells BigQuery to split the data into training and evaluation
sets for us - the evaluation set will be used later:

sql
CREATE OR REPLACE MODEL `taxi-rides-ny.nytaxi.tip_model`
OPTIONS
(model_type='linear_reg',
input_label_cols=['tip_amount'],
DATA_SPLIT_METHOD='AUTO_SPLIT') AS
SELECT
*
FROM
`taxi-rides-ny.nytaxi.yellow_tripdata_ml`
WHERE
tip_amount IS NOT NULL;

Training took around five minutes. Opening the model in the Explorer shows
what BigQuery did: the model type is linear regression, and it trained on a
temporary training dataset with a temporary evaluation dataset. The
evaluation tab shows the error metrics of the training run - the mean
squared error is around 8 and the mean absolute error is around 1:

That is not very optimal, but for a simple model and dataset it is perfectly
fine - the aim here is the workflow, not the most optimal model.

Checking the features

ML.FEATURE_INFO shows how BigQuery sees each feature:

sql
SELECT * FROM ML.FEATURE_INFO(MODEL `taxi-rides-ny.nytaxi.tip_model`);

The numeric columns - passenger count, trip distance, fare amount, tolls
amount - come with min, max and mean values, which are used for
standardization. The three columns we cast to strings - pickup location,
dropoff location and payment type - show up as categorical features with
their category counts.

Evaluating the model

ML.EVALUATE scores the model against a dataset:

sql
SELECT
*
FROM
ML.EVALUATE(MODEL `taxi-rides-ny.nytaxi.tip_model`,
(
SELECT
*
FROM
`taxi-rides-ny.nytaxi.yellow_tripdata_ml`
WHERE
tip_amount IS NOT NULL
));

Here we see the mean absolute error is 1 and the mean squared error is
around 150. These evaluation metrics are what we would use for optimizing
the model later on.

Predicting

ML.PREDICT applies the model to the dataset and adds the prediction as an
extra column:

sql
SELECT
*
FROM
ML.PREDICT(MODEL `taxi-rides-ny.nytaxi.tip_model`,
(
SELECT
*
FROM
`taxi-rides-ny.nytaxi.yellow_tripdata_ml`
WHERE
tip_amount IS NOT NULL
));

Every row now carries a predicted_tip_amount column next to the actual
tip_amount. With both side by side, we can also do a manual evaluation of
the model if we want to.

Explaining predictions

BigQuery can also tell us which features drive a prediction.
ML.EXPLAIN_PREDICT with top_k_features set to 3 lists the top three
features used for each prediction - you could just as well ask for the top
four:

sql
SELECT
*
FROM
ML.EXPLAIN_PREDICT(MODEL `taxi-rides-ny.nytaxi.tip_model`,
(
SELECT
*
FROM
`taxi-rides-ny.nytaxi.yellow_tripdata_ml`
WHERE
tip_amount IS NOT NULL
), STRUCT(3 as top_k_features));

Looking at the top three features, all our categorical features are there -
pickup location, dropoff location and payment type are what the model relies
on most, and the same pattern repeats in the second row and so on.

Hyperparameter tuning

We saw the model was not optimal. The standard way to improve it is
hyperparameter tuning. For linear regression, BigQuery lets us run several
trials (num_trials), execute some of them in parallel so the query runs
faster (max_parallel_trials), and try a range and candidate values for the
L1 and L2 regularization hyperparameters:

sql
CREATE OR REPLACE MODEL `taxi-rides-ny.nytaxi.tip_hyperparam_model`
OPTIONS
(model_type='linear_reg',
input_label_cols=['tip_amount'],
DATA_SPLIT_METHOD='AUTO_SPLIT',
num_trials=5,
max_parallel_trials=2,
l1_reg=hparam_range(0, 20),
l2_reg=hparam_candidates([0, 0.1, 1, 10])) AS
SELECT
*
FROM
`taxi-rides-ny.nytaxi.yellow_tripdata_ml`
WHERE
tip_amount IS NOT NULL;

This is a rich set. The same CREATE MODEL documentation lists many more
options - learning rate strategy, early stopping, minimum relative progress,
the learning rate itself. If you are familiar with machine learning, you can
use all these parameters to tune your model.

That is the full loop in BigQuery: the model is built, evaluated, explained
and tuned, all in SQL. In the
next unit we export it and serve
it from a Docker container.

Materials

  • big_query_ml.sql — building the ML feature table,
    creating and evaluating a linear_reg tip model, predicting, explaining
    predictions, and the hyperparameter-tuning variant.

Reference

File: 03-data-warehouse/06-deploying-a-machine-learning-model.md

video_url: https://www.youtube.com/watch?v=BjARzEWaznU prev_url: 05-machine-learning-in-bigquery.md next_url: ../04-analytics-engineering/01-analytics-engineering-basics.md

Deploying a Machine Learning Model from BigQuery

In the previous unit we trained a model
in BigQuery that predicts taxi tips. Here we take that model out of BigQuery:
we export it to Cloud Storage, serve it locally with TensorFlow Serving in a
Docker container, and call it over HTTP to get predictions.

The steps follow Google's export model
tutorial.

Exporting the model to Cloud Storage

After gcloud auth login (already done in the video), we use the bq
command-line tool to extract the model into a Cloud Storage bucket:

bash
bq --project_id taxi-rides-ny extract -m nytaxi.tip_model gs://taxi_ml_model/tip_model

Once the export finishes, the tip_model folder shows up in the
taxi_ml_model bucket:

Copying the model locally

Next we pull the model down to our machine with gsutil:

bash
mkdir /tmp/model
gsutil cp -r gs://taxi_ml_model/tip_model /tmp/model

The copy output shows what an exported BigQuery model is made of - it is a
TensorFlow model: assets, variables, and a few metadata files:

Serving the model with Docker

TensorFlow Serving expects a particular layout: a directory named after the
model, with one subdirectory per version, numbered. So we create
serving_dir/tip_model/1 and copy the model files into the version folder.
The serving directory can live anywhere - in the project or in a temp
directory:

bash
mkdir -p serving_dir/tip_model/1
cp -r /tmp/model/tip_model/* serving_dir/tip_model/1
docker pull tensorflow/serving
docker run -p 8501:8501 \
  --mount type=bind,source=`pwd`/serving_dir/tip_model,target=/models/tip_model \
  -e MODEL_NAME=tip_model -t tensorflow/serving &

This maps port 8501 of the container to port 8501 on our machine, mounts our
serving directory into the container, and tells TensorFlow Serving which
model to serve through the MODEL_NAME environment variable.

With docker ps we can check that the container is running:

Checking the model

TensorFlow Serving exposes a REST API. The model metadata lives at
http://localhost:8501/v1/models/tip_model - a GET there (in the video, with
Postman) should tell us the model is fine:

Indeed, the response says the tip model, version 1, is available - no error.

Predicting over HTTP

Now the actual prediction. We POST a JSON body with the same columns the
model was trained on - passenger count, trip distance, pickup and dropoff
location IDs, payment type, fare amount and tolls amount - to
http://localhost:8501/v1/models/tip_model:predict:

bash
curl -d '{"instances": [{"passenger_count":1, "trip_distance":12.2, "PULocationID":"193", "DOLocationID":"264", "payment_type":"1","fare_amount":20.4,"tolls_amount":0.0}]}' \
  -X POST http://localhost:8501/v1/models/tip_model:predict

For this ride, the model predicts a tip of around 3.2 dollars:

If we change the payment type to 2 and send the request again, the predicted
tip amount goes drastically down - to about 0.26 dollars:

And that is the whole loop: a model trained with SQL inside BigQuery,
exported, and served as a REST service from a Docker container on our own
machine.

File: 03-data-warehouse/module.yaml

schema_version: 2
content_id: c38e12d7-b52c-4e98-9dd5-441daff9ddf5
title: "Data Warehousing"
summary: "Query and optimize analytical data in BigQuery with partitioning, clustering, and built-in ML."
units:

  • content_id: 9e208dd7-f9b5-45d5-8cbf-908663e55a72
    title: "Data Warehouse and BigQuery"
    path: 01-data-warehouse-and-bigquery.md
  • content_id: dfc3ff3a-0071-446c-b2d1-40a02d22bf3b
    title: "Partitioning vs Clustering"
    path: 02-partitioning-vs-clustering.md
  • content_id: cd41db3e-4fdb-41e0-a73a-5d3159ec5ff5
    title: "BigQuery Best Practices"
    path: 03-bigquery-best-practices.md
  • content_id: 4cf6c6e0-6311-4f2e-9e0b-0bfc7202c014
    title: "Internals of BigQuery"
    path: 04-internals-of-bigquery.md
  • content_id: 2a33a931-88f1-49aa-8e6a-0163e97fb099
    title: "Machine Learning in BigQuery"
    path: 05-machine-learning-in-bigquery.md
  • content_id: 40110ed3-dd06-41d4-b1a5-63961af7110f
    title: "Deploying a Machine Learning Model from BigQuery"
    path: 06-deploying-a-machine-learning-model.md

File: 03-data-warehouse/extras/README.md

Quick hack to load files directly to GCS, without Airflow. Downloads csv files from https://nyc-tlc.s3.amazonaws.com/trip+data/ and uploads them to your Cloud Storage Account as parquet files.

  1. Install pre-reqs with uv sync
  2. Run: uv run python web_to_gcs_with_progress_bar.py
  3. or Run: uv run python web_to_gcs.py for less verbose (if you have fast internet connection in upload)

File: 03-data-warehouse/images/01-data-warehouse-and-bigquery-02-data-warehouse-diagram.json

{
"title": "Data warehouse architecture",
"description": "Data warehouse architecture: sources, staging area, warehouse, data marts, users.",
"canvas": {"width": 1380, "height": 280, "background": "#ffffff"},
"layout": {"type": "manual", "card_width": 220, "card_height": 100},
"nodes": [
{"id": "sources", "title": "Sources", "subtitle": "operational DBs · flat files", "color": "blue", "icon": "database", "x": 40, "y": 90},
{"id": "staging", "title": "Staging area", "subtitle": "clean and combine", "color": "amber", "icon": "settings", "x": 310, "y": 90},
{"id": "warehouse", "title": "Warehouse", "subtitle": "raw · summary · metadata", "color": "purple", "icon": "volume", "x": 580, "y": 90},
{"id": "marts", "title": "Data marts", "subtitle": "sales · purchasing · inventory", "color": "green", "icon": "database", "x": 850, "y": 90},
{"id": "users", "title": "Users", "subtitle": "analysis · reporting · mining", "color": "blue", "icon": "user", "x": 1120, "y": 90}
],
"edges": [
{"from": "sources", "to": "staging"},
{"from": "staging", "to": "warehouse"},
{"from": "warehouse", "to": "marts"},
{"from": "marts", "to": "users"}
]
}

File: 03-data-warehouse/images/04-internals-of-bigquery-01-architecture.json

{
"title": "BigQuery architecture",
"description": "BigQuery architecture: clients talk to a client interface, below it the Dremel query execution tree running on Borg, connected over the Jupiter network to Colossus storage.",
"canvas": {
"width": 720,
"height": 800,
"background": "#ffffff"
},
"layout": {
"type": "manual",
"card_width": 240,
"card_height": 100
},
"nodes": [
{
"id": "clients",
"title": "Clients",
"subtitle": "REST API \u00b7 Web UI \u00b7 bq CLI",
"color": "blue",
"icon": "browser",
"x": 240,
"y": 30
},
{
"id": "interface",
"title": "Client interface",
"subtitle": "public query endpoint",
"color": "purple",
"icon": "api",
"x": 240,
"y": 170
},
{
"id": "borg",
"variant": "boundary",
"title": "Borg",
"contains": [
"dremel"
]
},
{
"id": "dremel",
"title": "Dremel",
"subtitle": "query execution tree",
"color": "amber",
"icon": "workflow",
"x": 240,
"y": 340
},
{
"id": "jupiter",
"title": "Jupiter",
"subtitle": "datacenter network",
"color": "purple",
"icon": "websocket",
"x": 240,
"y": 480
},
{
"id": "colossus",
"title": "Colossus",
"subtitle": "distributed storage",
"color": "green",
"icon": "database",
"x": 240,
"y": 620
}
],
"edges": [
{
"from": "clients",
"to": "interface",
"bidirectional": true
},
{
"from": "interface",
"to": "dremel"
},
{
"from": "dremel",
"to": "jupiter",
"bidirectional": true
},
{
"from": "jupiter",
"to": "colossus",
"bidirectional": true
}
]
}

Explore Related Technologies & Topics

View All 9,111+ Repositories →

Discover similar high-velocity repositories, agent skills, and OpenAPI specifications across the ecosystem.

#course Browse course catalog → #data-engineering Browse data-engineering catalog → #dbt Browse dbt catalog → #docker Browse docker catalog → #free Browse free catalog → #kafka Browse kafka catalog → ⚡ MCP Servers Model Context Protocol → 🧠 Agent Skills Prompt & rule files → 🔌 OpenAPI Specs REST & Swagger schemas →