We're sorry but this app doesn't work properly without JavaScript enabled. Please enable it to continue.

This lesson's interactive features are locked, please to keep using them

Building an Integration Pipeline

Real integration pipelines typically follow the same pattern:

  1. Load raw files
  2. Standardize the schemas
  3. Combine similar datasets
  4. Enrich with lookup tables
  5. Validate the result

These steps represent the merging and concatenation tools we've been using in this chapter, arranged in a safe order:

  • Standardize first so that similar datasets can be stacked cleanly
  • Combine rows, then merge lookup tables to build one enriched fact table
  • Validate before calculating metrics so you don't build analysis on broken joins

Below is a small pipeline for support ticket data from two systems, plus agent and team reference tables. The goal is to combine the raw ticket exports, enrich them with context, and verify that the enrichment didn't duplicate rows.

Complete Pipeline

 
def create_ticket_dataset(data_dir: str) -> pd.DataFrame:
    # 1. Load all files
    email = pd.read_csv(f"{data_dir}/email_tickets.csv")
    chat = pd.read_csv(f"{data_dir}/chat_tickets.csv")
    agents = pd.read_csv(f"{data_dir}/agents.csv")
    teams = pd.read_csv(f"{data_dir}/teams.csv")

    # 2. Standardize columns
    email_std = email.rename(columns={"inbox": "queue"})
    chat_std = chat.rename(
        columns={
            "conversation_id": "ticket_id",
            "started_at": "opened_at",
        }
    )

    # 3. Add source identifiers
    email_std["source"] = "email"
    chat_std["source"] = "chat"

    # 4. Combine
    combined = pd.concat([email_std, chat_std], ignore_index=True)

    # 5. Enrich with agent information
    with_agents = pd.merge(combined, agents, on="agent_id", how="left")
    if len(with_agents) != len(combined):
        raise ValueError("agent merge changed row count")

    # 6. Enrich with teams
    complete = pd.merge(with_agents, teams, on="team_id", how="left")
    if len(complete) != len(with_agents):
        raise ValueError("team merge changed row count")

    # 7. Calculate metrics
    complete["response_hours"] = complete["response_minutes"] / 60

    return complete

That's the whole integration flow in one function. It starts with raw files and returns a clean, analysis-ready DataFrame.

Validation

Always validate after integration. For this pipeline, there are two especially important checks. First, row-count checks catch accidental duplication:

email = pd.read_csv(f"{data_dir}/email_tickets.csv")
chat = pd.read_csv(f"{data_dir}/chat_tickets.csv")
unified = create_ticket_dataset(data_dir)

original_count = len(email) + len(chat)
final_count = len(unified)

if original_count != final_count:
    print(f"ERROR: Expected {original_count}, got {final_count}")

Missing-value checks catch failed lookups:

missing_agents = unified["agent_email"].isna().sum()
missing_teams = unified["team_name"].isna().sum()

if missing_agents > 0:
    print(f"WARNING: {missing_agents} tickets missing agent data")

if missing_teams > 0:
    print(f"WARNING: {missing_teams} tickets missing team data")

Assignment

SnackStack's revenue operations team needs a pipeline that enriches sales with product and customer data. The result should be clean enough to analyze high-value orders without accidentally duplicating rows.

Complete the create_integration_pipeline function. It accepts three DataFrames: sales, products, and customers. It should enrich the sales data with product and customer details, with validation at each merge step.