The most insidious data problems aren't the ones that break your pipeline; they're the subtle shifts in data quality that go unnoticed, slowly eroding trust and leading to flawed insights. Imagine relying on an external RSS feed for critical market intelligence, only to discover weeks later that the `link` field stopped being a valid URL, or the `published` date format changed subtly, throwing off your time-series analysis. For data engineers and scientists building robust systems, especially those ingesting dynamic external data, manual quality checks are a losing battle. This post will walk you through setting up Great Expectations, a powerful data quality framework, to automatically validate an RSS feed from the Stripe Blog. You'll learn how to define expectations, including custom ones, to catch these silent data killers, significantly improving the integrity of your data pipelines and giving you confidence in the data you feed your models and dashboards.
Key Takeaways
- Great Expectations provides a structured, declarative way to define data quality expectations, turning implicit assumptions into explicit, testable assertions.
- Automating data quality checks at ingest points, especially for external APIs or feeds, is critical for preventing downstream data integrity issues.
- Custom expectations extend Great Expectations' capabilities, allowing you to define domain-specific validation rules beyond its rich built-in library.
- Data Docs generated by Great Expectations offer a valuable, human-readable report on data quality, fostering collaboration and understanding across teams.
- Proactive validation reduces debugging time, improves data reliability, and ensures that data-driven decisions are based on accurate information.
The Problem: Unreliable External Data
Our systems often depend on data from external sources – third-party APIs, public datasets, or, in this case, an RSS feed. While these sources can be incredibly valuable, they are also prone to unexpected changes: a field might disappear, its data type could shift, or its content might no longer adhere to expected patterns. When we built our internal market intelligence dashboard, pulling daily updates from the Stripe Blog's RSS feed, we initially relied on basic schema validation. This proved insufficient. A few weeks in, we noticed a significant drop in our ability to categorize posts by publication date, only to find the `published` field had started including timezone information inconsistently, breaking our parsing logic. We needed a systematic way to assert the *quality* of the data, not just its structure.
Data and Sources
For this tutorial, we'll be working with the Stripe Blog's RSS feed. This is a live, constantly updating data source, making it an excellent candidate for demonstrating how Great Expectations can monitor data quality in a dynamic environment.
Data accessed on 2024-07-20.
Step 1 — Installing Great Expectations and Initializing a Data Context
Before we can define any expectations, we need to set up our Great Expectations environment. This involves installing the library and initializing a Data Context, which is the central configuration for all your data quality work. It creates a `great_expectations` directory where all your configurations, expectation suites, and data docs will live.
The sub-problem here is setting up the foundational structure. We need to tell Great Expectations where to store its metadata.
import os
import great_expectations as gx
# Create a directory for our project if it doesn't exist
project_dir = "stripe_blog_quality"
if not os.path.exists(project_dir):
os.makedirs(project_dir)
os.chdir(project_dir)
# Initialize Great Expectations Data Context
# This creates the 'great_expectations' directory and its initial config
if not os.path.exists("great_expectations"):
context = gx.get_context()
context.convert_to_file_context()
print("Great Expectations Data Context initialized.")
else:
context = gx.get_context()
print("Great Expectations Data Context already exists.")
# Go back to parent directory for subsequent steps if needed,
# though we'll mostly operate within the context's scope.
os.chdir("..")
This snippet first ensures we're operating within a dedicated project directory. Then, it uses `gx.get_context()` to either load an existing context or initialize a new one. `context.convert_to_file_context()` ensures the context is stored on disk, making it persistent. This is the starting point for all data quality efforts.
Step 2 — Loading Data and Configuring a Data Source
With the Data Context initialized, the next challenge is to bring our data into Great Expectations. Great Expectations needs to know *where* your data is and *how* to access it. For our RSS feed, this means fetching it, parsing it into a structured format (like a Pandas DataFrame), and then configuring a data source within Great Expectations that points to this DataFrame.
The sub-problem is making our dynamic, external RSS data accessible and understandable by Great Expectations. We'll use `feedparser` to fetch and parse the XML, then `pandas` to convert it into a tabular structure.
import feedparser
import pandas as pd
import requests
def load_stripe_rss_feed(url: str = 'https://stripe.com/blog/feed.rss') -> pd.DataFrame:
"""Fetches and parses the Stripe Blog RSS feed into a Pandas DataFrame."""
try:
response = requests.get(url, timeout=10)
response.raise_for_status() # Raise an exception for HTTP errors
feed = feedparser.parse(response.text)
if feed.bozo:
print(f"Warning: RSS feed parsing had issues: {feed.bozo_exception}")
entries_data = []
for entry in feed.entries:
entries_data.append({
'title': entry.get('title'),
'link': entry.get('link'),
'published': entry.get('published'),
'summary': entry.get('summary'),
'author': entry.get('author')
})
if not entries_data:
print("Warning: No entries found in the RSS feed.")
return pd.DataFrame()
df = pd.DataFrame(entries_data)
return df
except requests.exceptions.RequestException as e:
print(f"Error fetching RSS feed from {url}: {e}")
return pd.DataFrame()
except Exception as e: # Catch other potential parsing errors
print(f"An unexpected error occurred during RSS parsing: {e}")
return pd.DataFrame()
# Inside our main script logic:
# df = load_stripe_rss_feed()
# if not df.empty:
# batch = context.add_or_update_datasource(
# name="stripe_blog_datasource",
# module_name="great_expectations.datasource",
# class_name="PandasDatasource"
# ).add_batch_definition(
# name="stripe_blog_feed",
# data_connector_name="default_runtime_data_connector",
# batch_identifiers={"pipeline_run_id": "manual_run"},
# data_asset_name="stripe_blog_posts"
# ).get_batch_from_data(df)
# print("Data Source and Batch created.")
The `load_stripe_rss_feed` function handles fetching the feed, parsing it, and converting it into a Pandas DataFrame, including basic error handling for network issues and parsing warnings. The commented-out Great Expectations code shows how we'd then add this DataFrame as a runtime batch to a Pandas Datasource. This "runtime" approach is perfect for dynamic data that isn't stored persistently in a database or file system for Great Expectations to connect to directly.
Step 3 — Defining Expectations and Implementing Custom Expectations
Now that we have our data, the core of data quality testing begins: defining expectations. These are assertions about your data that Great Expectations will validate. We'll start with some common built-in expectations and then tackle a more advanced scenario by implementing a custom expectation to check for specific keywords in titles, demonstrating how to extend Great Expectations for unique domain needs.
The sub-problem here is codifying our assumptions about the RSS feed data's quality and structure, and then extending that to cover business-specific logic.
from great_expectations.core import ExpectationConfiguration
from great_expectations.execution_engine import PandasExecutionEngine
from great_expectations.expectations.expectation import ColumnMapExpectation
from great_expectations.expectations.metrics import (
ColumnMapMetricProvider, column_map_metric
)
# 1. Define Standard Expectations
def build_stripe_rss_expectation_suite(context):
suite_name = "stripe_blog_rss_feed_suite"
if suite_name not in context.list_expectation_suite_names():
suite = context.create_expectation_suite(suite_name)
else:
suite = context.get_expectation_suite(suite_name)
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_to_exist",
kwargs={"column": "title"}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_to_exist",
kwargs={"column": "link"}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_to_exist",
kwargs={"column": "published"}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_values_to_not_be_null",
kwargs={"column": "title"}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_values_to_be_of_type",
kwargs={"column": "link", "type_lookup": "url"} # Check if it's a valid URL string
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_values_to_match_regex",
kwargs={"column": "published", "regex": r"^\w{3}, \d{2} \w{3} \d{4} \d{2}:\d{2}:\d{2} [+-]\d{4}$"}
# Example regex for 'Thu, 18 Jul 2024 10:00:00 +0000' format
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_value_lengths_to_be_between",
kwargs={"column
