Stock Market Pipeline with Prefect

Build an ELT pipeline that loads live stock data into BigQuery.

Introduction

30 Second Summary

Stock prices change every second. The professionals tracking them build automated pipelines that pull fresh data, clean it, and store it where anyone on the team can query it instantly.

In this project, you will build a stock market data pipeline that extracts live financial data from an API, transforms it into analysis-ready formats, and loads it into Google BigQuery where you can query it with SQL.

What You'll Build

You will run your pipeline and watch live stock market data flow from an API through transformation steps into a cloud warehouse, all tracked in a visual monitoring dashboard in your browser.

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

  • A running ELT pipeline that extracts live stock market data from the Strasmore API and loads it into Google BigQuery on command.
  • A Prefect orchestration dashboard where you can watch each pipeline task succeed, trigger re-runs, and see automatic retries kick in when something goes wrong.
  • SQL queries against your warehouse that answer questions like "which stocks have the highest dividend yields today?"
  • Secret Mission: Extend your pipeline to pull from three different data sources with data quality validation that skips bad data automatically.

Are there any prerequisites?

This project assumes you are comfortable with Python fundamentals, basic SQL, and navigating the terminal. No prior experience with data pipelines or cloud warehouses is needed.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters.

You're going to build an automated data pipeline that pulls live stock market data, transforms it, and loads it into a cloud warehouse. Understanding why automation matters here (instead of running a script once by hand) is the foundation of everything that follows.

Set Up Your Tools and Cloud Environment

A data pipeline needs two foundations: libraries that do the work locally, and a cloud warehouse that stores the results.

In this step you will create a Python environment with the pipeline libraries, install the Google Cloud SDK to authenticate with Google services, and set up a free BigQuery Sandbox project where your stock market data will land.

In this step, get ready to:
  • Create a Python virtual environment and install pipeline libraries.
  • Install and configure the Google Cloud SDK.
  • Set up a BigQuery Sandbox project with a dataset.
Create a Python virtual environment and install packages

A virtual environment keeps your pipeline's dependencies isolated from your system Python. This prevents version conflicts between projects.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir stock-pipeline
cd stock-pipeline
  • Create and activate a virtual environment by running these commands:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

Your terminal prompt now shows (venv) at the beginning, confirming the virtual environment is active.

  • Install the pipeline packages by running:
pip install prefect google-cloud-bigquery pandas pyarrow httpx

What are these packages?

  • prefect orchestrates your pipeline with retries, scheduling, and monitoring.
  • google-cloud-bigquery is the Python client for loading data into BigQuery.
  • pandas handles data transformation in DataFrames.
  • pyarrow serializes DataFrames to Parquet format for BigQuery loading.
  • httpx makes HTTP requests to the stock market API.

The install takes a minute or two. You will see a stream of download progress bars followed by a success message.

  • Confirm Prefect installed correctly by running:
prefect version

You should see a version number printed (e.g. 3.7.8).

  • Confirm the BigQuery client installed correctly by running:
python -c "from google.cloud import bigquery; print(bigquery.__version__)"

You should see a version number printed (e.g. 3.42.2).

Seeing ModuleNotFoundError?

Make sure your terminal prompt starts with (venv). If it does not, re-run the activate command from above. Packages install into the virtual environment, so they are only available when it is active.

Help me troubleshoot

Install Google Cloud SDK and authenticate

The Google Cloud SDK provides the gcloud command-line tool. Your pipeline uses it to authenticate with BigQuery so it can load data without managing service account keys manually.

  • Check whether gcloud is already installed by running:
gcloud --version

✔️ I see version info

The gcloud CLI is already installed. Skip ahead to the authentication step below.

ⓧ Command not found

You need to install the Google Cloud SDK. Follow the instructions for your operating system below.

macOS

The simplest method on macOS uses Homebrew.

  • Install the gcloud CLI by running:
brew update && brew install --cask google-cloud-sdk

After the install finishes, close your terminal and open a new one so the PATH updates take effect. Re-activate your virtual environment by running:

cd ~/Desktop/stock-pipeline
source venv/bin/activate
Don't have Homebrew?

Visit https://brew.sh and follow the one-line install command first, then retry the brew install command above.

Windows

Windows uses Google's official installer.

  • Open PowerShell and download the installer by running:
(New-Object Net.WebClient).DownloadFile("https://dl.google.com/dl/cloudsdk/channels/rapid/GoogleCloudSDKInstaller.exe", "$env:Temp\GoogleCloudSDKInstaller.exe")
& $env:Temp\GoogleCloudSDKInstaller.exe
  • Follow the installer prompts. Leave the default options selected (including bundled Python).
  • When the installer finishes, uncheck the option to start the Google Cloud CLI shell. You will use your existing terminal instead.
  • Close your terminal and open a new one so the PATH updates take effect.
  • Re-activate your virtual environment by running:
cd ~\Desktop\stock-pipeline
venv\Scripts\activate
  • Verify the install succeeded by running:
gcloud --version

You should now see version info printed.

  • Initialize a new gcloud configuration by running:
gcloud init

What will gcloud init ask?

The command launches an interactive setup. It will ask you to:

  • Log in with your Google account (opens a browser window).
  • Select or create a Google Cloud project. Choose the project you will create in the next substep, or create one now (e.g. stock-pipeline).
  • Optionally set a default region. You can skip this for now.
  • Set up Application Default Credentials so your Python code can authenticate with BigQuery by running:
gcloud auth application-default login

A browser window opens. Sign in with the same Google account you used during gcloud init. After you approve access, the terminal prints a confirmation that credentials have been saved.

What are Application Default Credentials?

Application Default Credentials (ADC) let libraries like google-cloud-bigquery authenticate automatically without you pasting keys into your code. The credentials are stored locally on your machine and picked up by any Google Cloud client library.

Browser did not open?

If no browser window appeared, the terminal will print a URL. Copy that URL, paste it into your browser manually, complete the sign-in flow, then copy the authorization code back into the terminal.

Help me complete the authentication flow

Set up BigQuery Sandbox

BigQuery is Google's serverless data warehouse. The Sandbox tier gives you 10 GB of storage and 1 TB of query processing per month at no cost and requires no credit card.

If you already created a project during gcloud init, select it from the project picker at the top of the page. Otherwise, create one now.

  • Click the project dropdown at the top of the console.
  • Click New Project.
  • For Project name, enter stock-pipeline (or a name of your choice).
  • For Organization, select No organization unless you are part of one.
  • Click Create.

You are redirected back to the BigQuery page. A sandbox notice confirms your project is active at no cost.

  • In the left Explorer pane, click the three-dot menu next to your project name.
  • Click Create dataset.
  • For Dataset ID, enter market_data.
  • For Location type, select Multi-region and choose US.
  • Click Create dataset.

The market_data dataset now appears under your project in the Explorer pane.

Why US multi-region?

Public datasets in BigQuery are stored in the US multi-region. Placing your own dataset in the same location avoids cross-region data movement costs and keeps queries fast.

Now run a quick test query to confirm everything connects properly.

  • Click the + button (or Compose a new query) in the query editor area.
  • Paste this SQL into the query editor:
SELECT name, number
FROM `bigquery-public-data.usa_names.usa_1910_current`
LIMIT 5;
  • Click Run.

Before you look at the results, predict what you expect to see. Five rows of names and numbers from a public dataset? Something else?

You should see a results table with 5 rows showing name and number columns. This confirms your BigQuery project and dataset are working.

Query failed or no results?

Make sure you selected the correct project from the project picker at the top of the console. The BigQuery API is enabled automatically for new projects, but if you are reusing an existing project you may need to enable it manually.

Help me fix my BigQuery query

Your local environment and cloud warehouse are ready. Next up, you will pull live stock market data from the Strasmore API and see it in your terminal.

Extract Stock Market Data from the API

Your environment is set up, your BigQuery warehouse is waiting, and your Python packages are ready to go. Now you need data to feed the pipeline.

Before you can transform or warehouse anything, you need raw data flowing into Python. In this step, you will pull live stock market data from the Strasmore API and prove you can see it locally as a structured table.

In this step, get ready to:
  • Write a Python script that calls the Strasmore API and receives live stock data.
  • Parse the API response and convert it into a pandas DataFrame you can work with.
Write an extraction function that calls the Strasmore API

The Strasmore API provides free stock market data with no API key or signup required. You send a GET request with an endpoint key, and the response comes back as JSON containing both metadata and the actual data rows.

  • Create a new file called pipeline.py in your project folder by running:
touch pipeline.py
  • Confirm the file exists by running:
ls

You should see pipeline.py listed in the output.

  • Open pipeline.py in VS Code and paste the following code:
from __future__ import annotations

from typing import Any

import httpx
import pandas as pd


STRASMORE_BASE_URL = "https://ai.strasmore.com/api/demo"


def extract(endpoint: str) -> dict[str, Any]:
    """Extract data from the Strasmore API for a given endpoint key."""
    url = f"{STRASMORE_BASE_URL}?q={endpoint}"
    print(f"Fetching {endpoint} from {url}")
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    data = response.json()
    print(f"Received {len(data.get('rows', []))} rows for {endpoint}")
    return data


if __name__ == "__main__":
    raw_data = extract("dividend_yield_leaders")
    print(raw_data.keys())

What does this code do?

  • STRASMORE_BASE_URL stores the base API address so you can reuse it for different endpoint keys later.
  • The extract function takes an endpoint key (like dividend_yield_leaders), builds the full URL, and makes a GET request using httpx.
  • response.raise_for_status() raises an error immediately if the API returns a failure code, rather than silently passing bad data forward.
  • The if __name__ block calls the function and prints the response keys so you can examine the JSON structure.
  • Save pipeline.py.
  • Run the script to see what the API sends back:
python pipeline.py

You should see output like this:

Fetching dividend_yield_leaders from https://ai.strasmore.com/api/demo?q=dividend_yield_leaders
Received 12 rows for dividend_yield_leaders
dict_keys(['key', 'label', 'nl', 'sql', 'columns', 'rows', 'source', 'more'])

What is in the API response?

The Strasmore API wraps each response in metadata alongside the actual data. Here is what each key means:

  • key and label identify the query you ran.
  • sql shows the exact SQL the API used to produce the data.
  • columns is a list of column names, and rows is a list of data rows. These two fields are what you need for your DataFrame.
  • source and more provide attribution and links back to Strasmore.

Seeing a connection error?

  • Make sure your virtual environment is activated. You should see a (venv) prefix in your terminal prompt.
  • Check your internet connection. The Strasmore API requires outbound HTTPS access.
  • If you see ModuleNotFoundError: No module named 'httpx', re-run pip install httpx inside your activated virtual environment.

Help me debug a connection error when calling the Strasmore API

Convert the API response into a pandas DataFrame

You now have raw JSON flowing into Python, but a dictionary of keys and row lists is hard to work with. A pandas DataFrame turns that raw data into a proper table you can filter, sort, and analyze.

  • In pipeline.py, add the following function between the extract function and the if __name__ block:
def transform_raw(raw_data: dict[str, Any]) -> pd.DataFrame:
    """Convert raw API response into a pandas DataFrame."""
    columns = raw_data["columns"]
    rows = raw_data["rows"]
    df = pd.DataFrame(rows, columns=columns)
    print(df)
    return df

What does this function do?

  • It pulls the columns and rows fields out of the API response, ignoring the metadata you do not need for analysis.
  • pd.DataFrame(rows, columns=columns) constructs a tabular DataFrame using the column headers from the API and the row data.
  • Printing the DataFrame lets you immediately see the data in a readable table format.
  • Now update the if __name__ block at the bottom of the file.
  • Find this line:
    print(raw_data.keys())
  • Replace it with a call to your new function, so the block looks like this:
if __name__ == "__main__":
    raw_data = extract("dividend_yield_leaders")
    df = transform_raw(raw_data)
  • Save pipeline.py.

Before you run this, what do you expect to see in the output? Think about what the API returns and what a DataFrame print looks like.

  • Run the updated script:
python pipeline.py

You should see a table of 12 stock tickers with their dividend yields, printed as a pandas DataFrame. The output includes columns like ticker, dividend_yield_pct, and as_of_label, confirming that live API data is flowing into your Python environment.

Not seeing a table of stocks?

  • If you see a NameError for transform_raw, make sure you placed the function definition above the if __name__ block, not below it.
  • If the DataFrame prints but shows all columns as object types, that is expected at this stage. You will fix the types in the next step.
  • Double check you removed the print(raw_data.keys()) line and replaced it with df = transform_raw(raw_data).

Help me debug why my DataFrame is not printing stock data

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and the output shows a proper DataFrame with stock tickers.

ⓧ I'd like to double check the full code

Here is the full pipeline.py at this point:

from __future__ import annotations

from typing import Any

import httpx
import pandas as pd


STRASMORE_BASE_URL = "https://ai.strasmore.com/api/demo"


def extract(endpoint: str) -> dict[str, Any]:
    """Extract data from the Strasmore API for a given endpoint key."""
    url = f"{STRASMORE_BASE_URL}?q={endpoint}"
    print(f"Fetching {endpoint} from {url}")
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    data = response.json()
    print(f"Received {len(data.get('rows', []))} rows for {endpoint}")
    return data
def transform_raw(raw_data: dict[str, Any]) -> pd.DataFrame:
    """Convert raw API response into a pandas DataFrame."""
    columns = raw_data["columns"]
    rows = raw_data["rows"]
    df = pd.DataFrame(rows, columns=columns)
    print(df)
    return df


if __name__ == "__main__":
    raw_data = extract("dividend_yield_leaders")
    df = transform_raw(raw_data)

You now have live stock market data flowing from an API into a structured pandas DataFrame. Next up, you will push this data into BigQuery and discover why loading raw data without proper types creates a real problem for analytics.

Load Raw Data into BigQuery

In the previous step, you pulled live stock market data from the Strasmore API and printed it locally as a DataFrame.

Seeing data in your terminal is useful for debugging, but it does not give you a queryable warehouse. In this step, you will push that raw DataFrame directly into BigQuery so you can query it with SQL. Loading data quickly proves the pipeline works end-to-end. It also reveals a critical data engineering problem: raw data loaded without proper types is analytically useless.

In this step, get ready to:
  • Write a load function that pushes a DataFrame into BigQuery.
  • Run the full extract-and-load script and verify data in the BigQuery console.
  • Discover that arithmetic queries fail on raw string-typed columns.
Write a load function that pushes the DataFrame to BigQuery

Your extraction script already pulls data and builds a DataFrame. Now you need a function that takes that DataFrame and writes it into a BigQuery table, replacing whatever was there before.

  • In pipeline.py, find the existing import block at the top of the file. Add the BigQuery import and a new constant by copying the following code below the import pandas as pd line:
from google.cloud import bigquery


STRASMORE_BASE_URL = "https://ai.strasmore.com/api/demo"
BIGQUERY_DATASET = "market_data"

What changed at the top of the file?

You added from google.cloud import bigquery to bring in the BigQuery client library.

You also replaced the standalone STRASMORE_BASE_URL line with two constants grouped together. BIGQUERY_DATASET stores the dataset name so you can reference it in the load function without hardcoding it again.

  • Now add the load function below the transform_raw function. Copy this code after the return df line of transform_raw:
def load(df: pd.DataFrame, table_name: str, project_id: str) -> None:
    """Load a DataFrame into BigQuery."""
    client = bigquery.Client(project=project_id)
    table_id = f"{project_id}.{BIGQUERY_DATASET}.{table_name}"

    job_config = bigquery.LoadJobConfig(
        write_disposition="WRITE_TRUNCATE",
    )

    job = client.load_table_from_dataframe(df, table_id, job_config=job_config)
    job.result()

    table = client.get_table(table_id)
    print(f"Loaded {table.num_rows} rows to {table_id}")

What does this code do?

  • bigquery.Client(project=project_id) creates a client that authenticates using the Application Default Credentials you set up earlier.
  • table_id follows BigQuery's fully qualified format: project.dataset.table.
  • WRITE_TRUNCATE means each run replaces the entire table with fresh data, so you always have the latest snapshot.
  • load_table_from_dataframe() serializes the DataFrame to Parquet (using pyarrow) and uploads it to BigQuery.
  • job.result() blocks until the load job completes, so the next line can safely read the table.

Next, update the __main__ block so it calls load after extracting and transforming.

  • In pipeline.py, scroll down to the if __name__ == "__main__": block at the bottom. Find these lines:
if __name__ == "__main__":
    raw_data = extract("dividend_yield_leaders")
    df = transform_raw(raw_data)
  • Add this line below df = transform_raw(raw_data), so the block now looks like this:
if __name__ == "__main__":
    raw_data = extract("dividend_yield_leaders")
    df = transform_raw(raw_data)
    load(df, "dividend_leaders", [[PROJECT_ID="your-project-id"]])

What is the project ID?

This is the Google Cloud project ID you created in Step 1 (e.g., stock-pipeline). You can find it in the Google Cloud Console at the top of the page, or by running gcloud config get-value project in your terminal.

  • Save pipeline.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py with the new import, constant, load function, and updated __main__ block.

ⓧ I'd like to double check the full code

Scroll up through pipeline.py and confirm these sections are present, in order:

  • Imports: from __future__ import annotations, from typing import Any, import httpx, import pandas as pd, from google.cloud import bigquery.
  • Constants: STRASMORE_BASE_URL and BIGQUERY_DATASET.
  • The extract function (unchanged from the previous step).
  • The transform_raw function (unchanged from the previous step).
  • The new load function with bigquery.Client, LoadJobConfig, load_table_from_dataframe, and the row count print.
  • The __main__ block calling extract, transform_raw, then load.
Run the full extract-and-load script, then query BigQuery

Your script now extracts data from the API and loads it straight into BigQuery. Time to run it and prove data lands in the warehouse.

  • Run the pipeline by executing this command in your terminal:
python pipeline.py

You should see output showing the fetched rows, the DataFrame print, and a final line confirming how many rows were loaded (e.g., Loaded 12 rows to your-project-id.market_data.dividend_leaders).

Seeing a permission or authentication error?

  • Make sure you ran gcloud auth application-default login in Step 1 and completed the browser sign-in.
  • Confirm your project ID matches the one in the Google Cloud Console. Run gcloud config get-value project to check.
  • If you see a pyarrow error, confirm you installed it: pip install pyarrow.

Help me debug my BigQuery authentication error when running pipeline.py

  • Open the BigQuery console in your browser.
  • Click on your project, then expand the market_data dataset in the left sidebar.
  • Click on the dividend_leaders table that just appeared.
  • Run this query to confirm data arrived:
SELECT * FROM market_data.dividend_leaders LIMIT 5

You should see five rows of stock data with columns like ticker, dividend_yield_pct, and others. The data is in BigQuery.

Now comes the key question. The dividend_yield_pct column looks like numbers. Can BigQuery actually do math on them?

Before you run the next query, predict: will BigQuery successfully add 1 to dividend_yield_pct? Or will something go wrong?

  • Run this arithmetic query in the BigQuery console:
SELECT ticker, dividend_yield_pct + 1 FROM market_data.dividend_leaders

The query fails. BigQuery returns an error indicating there is no matching signature for the + operator with STRING and INT64 argument types.

Why did this fail?

When you loaded the DataFrame without specifying types, pandas serialized every column from the API as a string. BigQuery stored dividend_yield_pct as a STRING column, even though the values look like numbers (e.g., "5.23").

BigQuery cannot perform arithmetic on strings. The + operator only works on numeric types like FLOAT64 or INT64. This is the core data engineering problem: raw data loads without transformation are not analysis-ready.

What is the ELT pattern?

You just experienced why the T in ELT (Extract, Load, Transform) exists. Loading data fast is great for proving connectivity, but you need a transformation step to cast types, handle nulls, and enforce schemas before the data is useful for analytics.

Your data is in BigQuery, but it is locked in string form. In the next step, you will build a proper transformation function that casts numeric columns to floats and parses dates, so your queries actually work.

Transform Data for Analytics

In the previous step, you loaded raw stock market data into BigQuery and hit a wall. Arithmetic queries on columns like dividend_yield_pct failed because every value was stored as a STRING.

This is the core data engineering problem: raw API data is never analysis-ready. In this step, you'll build the "T" in your ELT pipeline by casting numeric columns to floats, parsing dates, and adding lineage metadata so BigQuery can actually do math on your data.

In this step, get ready to:
  • Build a transformation function that casts strings to proper numeric and datetime types.
  • Reload the typed data into BigQuery and verify arithmetic queries work.
Build a transformation function

The API returns every value as a string. Your transformation function needs to identify which columns are numeric, cast them to floats, parse any date columns, and stamp each row with a loaded_at timestamp for pipeline lineage tracking.

  • In pipeline.py, find the transform_raw function and delete it entirely.
  • Replace it with the following transform function:
def transform(raw_data: dict[str, Any]) -> pd.DataFrame:
    """Transform raw API response into a typed DataFrame."""
    columns = raw_data["columns"]
    rows = raw_data["rows"]

    df = pd.DataFrame(rows, columns=columns)

    # Cast numeric columns from string to float
    for col in df.columns:
        if col == "ticker" or col == "as_of":
            continue
        df[col] = pd.to_numeric(df[col], errors="coerce")

    # Parse date columns if present
    if "as_of" in df.columns:
        df["as_of"] = pd.to_datetime(df["as_of"], errors="coerce")

    # Add pipeline metadata
    df["loaded_at"] = pd.Timestamp.now(tz="UTC")

    print(f"Transformed DataFrame: {len(df)} rows, {len(df.columns)} columns")
    print(f"Dtypes:\n{df.dtypes}")
    return df

What does this code do?

  • The for col in df.columns loop skips ticker (a label, not a number) and as_of (a date, not a number). Everything else gets cast to a float.
  • pd.to_numeric(errors="coerce") converts values that cannot be parsed into NaN instead of crashing the script. This handles any unexpected non-numeric values gracefully.
  • pd.to_datetime(errors="coerce") parses date strings into proper datetime objects. Unparseable values become NaT (Not a Time).
  • The loaded_at column records the exact UTC time the pipeline ran. This is pipeline lineage metadata that tells you when each row was ingested.

Now you need to update the __main__ block to call your new function instead of the deleted one.

  • Scroll down to the if __name__ == "__main__": block at the bottom of pipeline.py.
  • Find this line:
    df = transform_raw(raw_data)
  • Replace transform_raw with transform, so the line now reads:
    df = transform(raw_data)
  • Also update the load function's docstring to reflect that it now receives typed data. Find:
    """Load a DataFrame into BigQuery."""
  • Replace it with:
    """Load a DataFrame into BigQuery with explicit schema."""
  • Save pipeline.py.
  • Run the updated pipeline by entering this command in your terminal:
python pipeline.py

You should see output showing the transformed DataFrame's row count and its dtypes. Columns like dividend_yield_pct now show float64 instead of object, and loaded_at shows as a timestamp.

Seeing a NameError for transform?

  • Make sure you deleted the old transform_raw function and replaced it with transform. The __main__ block must call transform, not transform_raw.
  • Check that the function definition uses def transform(raw_data: dict[str, Any]) with the exact name transform.

Still stuck? Why is my transform function throwing a NameError?

Verify the fix in BigQuery

Your pipeline just reloaded the data with proper types. Because the DataFrame now has float64 and datetime64 dtypes, load_table_from_dataframe serializes it to Parquet format, which carries the types into BigQuery automatically.

Before you run the arithmetic query again, predict: do you think it will succeed this time, or fail the same way it did before?

  • Switch back to the BigQuery console in your browser.
  • Run the following query in the query editor:
SELECT ticker, dividend_yield_pct + 1 AS adjusted
FROM market_data.dividend_leaders
ORDER BY dividend_yield_pct DESC
LIMIT 5

The query returns five rows with numeric results in the adjusted column. Each value is the original dividend_yield_pct plus 1, proving BigQuery now treats the column as a FLOAT, not a STRING.

Why did this fix work?

When you loaded raw data in the previous step, the DataFrame had object dtype on every column (pandas' catch-all for strings). BigQuery inferred all columns as STRING.

Now your transform function explicitly casts numeric columns to float64. When load_table_from_dataframe serializes the DataFrame, pyarrow preserves those types in the Parquet payload. BigQuery reads the Parquet schema and creates matching FLOAT64 columns.

Query still failing with a type error?

  • Confirm the pipeline re-ran successfully. Look for the Loaded X rows message in your terminal output.
  • Check that your LoadJobConfig uses WRITE_TRUNCATE. If it appends instead of replacing, the old STRING-typed schema persists.
  • In the BigQuery console, click on the dividend_leaders table and check the Schema tab. dividend_yield_pct should show as FLOAT, not STRING.

Need help debugging? Why is my BigQuery column still showing as STRING after reloading?

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and the query returned numeric results.

ⓧ I'd like to double check the full code

Here is the full pipeline.py at this point. Compare yours against it:

from __future__ import annotations

from typing import Any

import httpx
import pandas as pd
from google.cloud import bigquery


STRASMORE_BASE_URL = "https://ai.strasmore.com/api/demo"
BIGQUERY_DATASET = "market_data"
def extract(endpoint: str) -> dict[str, Any]:
    """Extract data from the Strasmore API for a given endpoint key."""
    url = f"{STRASMORE_BASE_URL}?q={endpoint}"
    print(f"Fetching {endpoint} from {url}")
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    data = response.json()
    print(f"Received {len(data.get('rows', []))} rows for {endpoint}")
    return data
def transform(raw_data: dict[str, Any]) -> pd.DataFrame:
    """Transform raw API response into a typed DataFrame."""
    columns = raw_data["columns"]
    rows = raw_data["rows"]

    df = pd.DataFrame(rows, columns=columns)

    for col in df.columns:
        if col == "ticker" or col == "as_of":
            continue
        df[col] = pd.to_numeric(df[col], errors="coerce")

    if "as_of" in df.columns:
        df["as_of"] = pd.to_datetime(df["as_of"], errors="coerce")

    df["loaded_at"] = pd.Timestamp.now(tz="UTC")

    print(f"Transformed DataFrame: {len(df)} rows, {len(df.columns)} columns")
    print(f"Dtypes:\n{df.dtypes}")
    return df
def load(df: pd.DataFrame, table_name: str, project_id: str) -> None:
    """Load a DataFrame into BigQuery with explicit schema."""
    client = bigquery.Client(project=project_id)
    table_id = f"{project_id}.{BIGQUERY_DATASET}.{table_name}"

    job_config = bigquery.LoadJobConfig(
        write_disposition="WRITE_TRUNCATE",
    )

    job = client.load_table_from_dataframe(df, table_id, job_config=job_config)
    job.result()

    table = client.get_table(table_id)
    print(f"Loaded {table.num_rows} rows to {table_id}")
if __name__ == "__main__":
    raw_data = extract("dividend_yield_leaders")
    df = transform(raw_data)
    load(df, "dividend_leaders", "your-project-id")

Your pipeline now extracts, transforms, and loads properly typed data into a cloud data warehouse. Next up, you'll orchestrate this entire flow with Prefect so it runs on a schedule with automatic retries and a monitoring dashboard.

Orchestrate with Prefect

Your pipeline extracts, transforms, and loads stock market data into BigQuery. But right now it only runs when you manually execute the script.

A production pipeline needs to run on a schedule, retry automatically when the API has a hiccup, and give you a dashboard where you can monitor every run. Prefect adds all three with just a few decorators and a single serve command.

In this step, get ready to:
  • Refactor your pipeline functions into Prefect tasks and a flow.
  • Add scheduling and serve a deployment that runs daily at 9 AM.
  • View your pipeline run in the Prefect monitoring dashboard.
Refactor the pipeline into Prefect tasks and a flow

Prefect uses two decorators to orchestrate your code. @task marks individual units of work that Prefect tracks, retries, and logs. @flow wraps the top-level function that coordinates those tasks.

  • In pipeline.py, find the import section at the top of the file. Locate this line:
from google.cloud import bigquery
  • Add this import directly below it:
from prefect import flow, task
  • Now find the extract function definition. Add the @task decorator directly above it so it looks like this:
@task(retries=3, retry_delay_seconds=[2, 5, 15], log_prints=True)
def extract(endpoint: str) -> dict[str, Any]:
  • Find the transform function definition. Add the @task decorator above it:
  • Find the load function definition. Add the @task decorator above it:
@task(log_prints=True)
def transform(raw_data: dict[str, Any]) -> pd.DataFrame:
@task(log_prints=True)
def load(df: pd.DataFrame, table_name: str, project_id: str) -> None:

What do these decorators do?

The @task decorator tells Prefect to track each function as a separate unit of work. You can see each task's status, duration, and logs individually in the Prefect UI.

The retries=3 on extract means if the API call fails, Prefect will automatically retry up to 3 times with increasing delays (2, 5, then 15 seconds). The log_prints=True setting captures your print() statements into Prefect's log system so they appear in the dashboard.

  • Now replace the entire if __name__ == "__main__": block at the bottom of pipeline.py with the flow function and a new main block:
@flow(name="stock_market_pipeline", log_prints=True)
def stock_market_pipeline(
    project_id: str = "your-project-id",
    endpoint: str = "dividend_yield_leaders",
    table_name: str = "dividend_leaders",
) -> None:
    """Extract stock market data, transform it, and load into BigQuery."""
    raw_data = extract(endpoint)
    df = transform(raw_data)
    load(df, table_name, project_id)
    print(f"Pipeline complete for {endpoint}")


if __name__ == "__main__":
    stock_market_pipeline(
        project_id="your-project-id",
        endpoint="dividend_yield_leaders",
        table_name="dividend_leaders",
    )

What does the flow function do?

The @flow decorator wraps stock_market_pipeline as the top-level orchestrator. It coordinates the three tasks in sequence: extract, then transform, then load.

Prefect automatically tracks the flow run's start time, end time, final state, and the relationship between the flow and its tasks. The name="stock_market_pipeline" parameter sets how this flow appears in the Prefect dashboard.

  • Replace your-project-id in both the function default and the __main__ call with your actual Google Cloud project ID.
  • Save pipeline.py.

Before you run this, what do you think will be different about the terminal output compared to before?

  • Run the pipeline by executing this command:
python pipeline.py

You should see timestamped log lines from Prefect showing the flow and each task being created and completed. The output looks something like this:

10:08:02.948 | INFO    | prefect.engine - Created flow run 'solid-gibbon' for flow 'stock-market-pipeline'
10:08:03.703 | INFO    | Flow run 'solid-gibbon' - Created task run 'extract-0' for task 'extract'
10:08:03.704 | INFO    | Flow run 'solid-gibbon' - Executing 'extract-0' immediately...
10:08:04.105 | INFO    | Task run 'extract-0' - Finished in state Completed()
...

What changed?

Before, you saw plain print() output. Now Prefect wraps every function call in tracked state transitions. Each task gets its own log context, and Prefect assigns your flow run a random name (like solid-gibbon) so you can identify it later in the dashboard.

Seeing an import error?

Make sure your virtual environment is active. The prefect package is only available inside the venv you created in Step 1.

If you see a BigQuery permissions error, double check that your-project-id matches your actual Google Cloud project ID exactly.

help me debug my Prefect pipeline import or authentication error

Add scheduling and serve the deployment

Your pipeline now runs with Prefect orchestration, but you still trigger it manually. A real data pipeline runs on a schedule automatically. Prefect's .serve() method creates a lightweight deployment that runs your flow on a cron schedule.

  • Create a new file called serve_pipeline.py in the same folder as pipeline.py by running:
touch serve_pipeline.py
  • Open serve_pipeline.py in VS Code and paste the following code:
from pipeline import stock_market_pipeline


if __name__ == "__main__":
    stock_market_pipeline.serve(
        name="stock-market-daily",
        cron="0 9 * * *",
        parameters={
            "project_id": "your-project-id",
            "endpoint": "dividend_yield_leaders",
            "table_name": "dividend_leaders",
        },
    )

What does this file do?

The .serve() method registers your flow as a deployment with a schedule. The cron="0 9 * * *" expression means "run at 9:00 AM every day". When this script is running, Prefect will automatically trigger a new pipeline run at the scheduled time.

The parameters dict passes the default values for each scheduled run. You can override these when triggering manual runs from the UI.

  • Replace your-project-id with your actual Google Cloud project ID.
  • Save serve_pipeline.py.

Now you need two terminal windows: one for the Prefect server and one to run your pipeline.

  • In your current terminal, start the Prefect server by running:
prefect server start

This starts a local Prefect API server and dashboard. The terminal will show the server running and display the UI URL. It takes over the terminal, so you need a second one to run commands.

  • Open a new terminal window.
  • Activate your virtual environment in the new terminal (the same source .venv/bin/activate or .venv\Scripts\activate command from Step 1).
  • Run the pipeline directly to confirm it works with the Prefect server by running:
python pipeline.py
  • Open your browser and navigate to http://localhost:4200.

You should see the Prefect dashboard with your flow run listed. Click on the flow run to see the individual tasks (extract, transform, load) with green success indicators.

What should I see in the Prefect UI?

The dashboard shows your flow run with a name like solid-gibbon and a green Completed state. Inside it, you can see each task run (extract, transform, load) with its own duration, state, and logs. This is the observability layer that makes your pipeline production-ready.

  • Back in your second terminal, trigger a manual run from the Prefect UI by clicking the **Quick run** button on your flow's page in the dashboard.

You should see a new flow run appear in the UI and complete with all tasks green. This confirms on-demand execution works alongside the scheduled deployment.

Dashboard not loading?

Check that the first terminal is still running prefect server start. If you accidentally closed it, restart it.

If http://localhost:4200 shows a connection error, wait 10 seconds for the server to finish starting up. You should see a line in the server terminal confirming it is ready.

help me troubleshoot my Prefect server or dashboard connection

✔️ Awesome, I've got everything!

Great. Double check you have saved both pipeline.py and serve_pipeline.py.

ⓧ I'd like to double check the full code

Scroll up and confirm each section is present in pipeline.py, in this order:

  • Imports including from prefect import flow, task.
  • Constants STRASMORE_BASE_URL and BIGQUERY_DATASET.
  • The extract function with @task(retries=3, retry_delay_seconds=[2, 5, 15], log_prints=True).
  • The transform function with @task(log_prints=True).
  • The load function with @task(log_prints=True).
  • The stock_market_pipeline function with @flow(name="stock_market_pipeline", log_prints=True).
  • The if __name__ == "__main__": block calling stock_market_pipeline().

For serve_pipeline.py, confirm it imports stock_market_pipeline from pipeline and calls .serve() with name="stock-market-daily" and cron="0 9 * * *".

Your pipeline is now fully orchestrated. You have automatic retries, a monitoring dashboard, and a schedule that will run your pipeline every morning. Next up, the Secret Mission challenges you to extend this to multiple data sources with quality validation.

Secret mission

Multi-Endpoint Pipeline with Data Quality Validation

Your pipeline handles one endpoint. Real production pipelines ingest from multiple sources and must validate data quality before loading. Can you extend your pipeline to pull from three different Strasmore endpoints, validate each dataset, and gracefully skip any that fail quality checks?

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses BigQuery Sandbox, which is free and requires no credit card. There are no ongoing costs.

Resources you used:

  • Google Cloud project with BigQuery Sandbox (free, no billing).
  • BigQuery dataset market_data with tables (dividend_leaders, biggest_movers, revenue_leaders).
  • Local Prefect server process.
  • Local project files (pipeline.py, pipeline_multi.py, serve_pipeline.py, virtual environment).
  • Application Default Credentials (local credential file).

Keep everything running

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

  • Your BigQuery project stays active at no cost. Data expires automatically after 60 days in sandbox mode.
  • The Prefect server continues running in your terminal. Press Ctrl+C when you want to stop it.
  • Your local project files and virtual environment remain on disk, ready for future work.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your BigQuery data and local files stay intact.

  • Stop the Prefect server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate your virtual environment by running:
  • Your BigQuery data remains queryable at no cost. No charges will occur while the project is idle.
deactivate

To resume later, reactivate your virtual environment and start the Prefect server again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is permanent.

  • Stop the Prefect server by pressing Ctrl+C in its terminal.
  • Revoke your Application Default Credentials by running:
gcloud auth application-default revoke
  • Delete your Google Cloud project. This removes BigQuery, the dataset, and all tables in one action. In the Google Cloud Console, go to the Manage resources page.
  • Select your project (e.g. stock-pipeline).
  • Click Delete.
  • Enter the project ID and click Shut down.

Delete your local project files by running the command for your operating system:

macOS

cd ~/Desktop
rm -rf stock-pipeline

Windows

cd ~/Desktop
rmdir /s /q stock-pipeline

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade stock market data pipeline that extracts live financial data, transforms it into analysis-ready tables, and loads it into a cloud data warehouse on a schedule.

You've learned how to:

  • Built a complete ELT pipeline that extracts live stock market data from an API, transforms raw JSON into typed DataFrames with proper schema enforcement, and loads clean data into Google BigQuery where you can query it with SQL.
  • Discovered the core data engineering problem firsthand: raw data loads without type coercion are analytically useless. You fixed it by casting numeric columns, parsing dates, and enforcing schema on load.
  • Orchestrated the entire workflow with Prefect, adding automatic retries, a cron schedule for daily runs, and a monitoring dashboard where you can track every task's success or failure in real time.
  • Secret Mission: Extended your pipeline to pull from three endpoints with data quality validation gates that skip loading when data fails checks.

Ready to quiz yourself?