Luigi developer on retainer: WeeklySalesReportPipeline ExtractSalesData changed to TSV on August 18 — ComputeSalesMetrics reads comma-delimited — df.get() returns 0 for all revenue columns — 7 weekly reports emailed with $0 revenue — Luigi reports SUCCESS

Luigi 3.5.1 • Python 3.11 • Amazon S3 (S3Target) • pandas 2.1.4 • AWS Glue • delimiter mismatch • silent pipeline failure • df.get() default • retainer billing

On October 6, 2026, the VP of Sales at Pinecrest Analytics Partners LLC opened a board preparation session reviewing Q3 performance and pulled the last seven weekly sales reports. Every single one showed weekly_revenue: $0, units_sold: 0, and avg_order_value: $0 — for every week from August 18 through September 29. The data engineering team was escalated to immediately. The Luigi developer retainer was engaged the same day. The cause was a two-line change deployed on August 18 that no automated system had flagged in 49 days: a developer had changed ExtractSalesData to write TSV instead of CSV to avoid commas in product names corrupting the output, but had not updated the downstream ComputeSalesMetrics task, which continued reading with pd.read_csv’s default comma delimiter. When pandas reads a TSV file using a comma as the delimiter, it produces a DataFrame with a single column whose name is the entire tab-separated header row and whose values are full tab-separated data rows. The numeric columns — weekly_revenue, units_sold, avg_order_value — do not exist in this malformed DataFrame. The task code used df.get('weekly_revenue', pd.Series([0])) to retrieve each column defensively, which silently returned the default pd.Series([0]) for every missing column, and .sum() on that series produced 0.0. The resulting metrics JSON was written to the S3Target output path, Luigi confirmed the file existed, marked ComputeSalesMetrics as complete(), and reported the entire WeeklySalesReportPipeline as SUCCESS. Seven weekly PDF reports were generated and emailed to six sales region managers, each showing $0 across all revenue metrics. No exception was raised. No warning was logged. No alert fired.

Pinecrest Analytics Partners LLC ran the WeeklySalesReportPipeline in pipelines/sales_reporting.py using Luigi 3.5.1 on Python 3.11. The pipeline extracted weekly sales data from an internal database into Amazon S3 using S3Target, computed aggregate metrics using pandas 2.1.4, and wrote the metrics to a JSON file that was then consumed by AWS Glue to generate a formatted weekly PDF report emailed to regional sales leadership every Monday. The pipeline had been running reliably since May 2026. On August 18, 2026, a developer on the team noticed that some product names contained commas — for example, “Widget, Deluxe Edition” — and that these commas were causing the CSV output to parse incorrectly when opened in Excel by non-technical stakeholders. The developer changed the writer in ExtractSalesData from csv.writer(f, delimiter=',') to csv.writer(f, delimiter='\t') and updated the S3 output path from .csv to .tsv. The downstream task ComputeSalesMetrics, which called pd.read_csv(target.path), was not updated. The change passed code review because the reviewer focused on the extraction logic and did not trace the output through to all downstream consumers:

# pipelines/sales_reporting.py
# Luigi 3.5.1, Python 3.11
# Change deployed August 18, 2026

class ExtractSalesData(luigi.Task):
    week_start = luigi.DateParameter()

    def output(self):
        return luigi.contrib.s3.S3Target(
            f's3://pinecrest-data/sales/weekly-{self.week_start}.tsv'
            # CHANGED: was .csv — updated to .tsv to match new delimiter
        )

    def run(self):
        rows = fetch_sales_data(self.week_start)  # returns list of dicts
        with self.output().open('w') as f:
            writer = csv.writer(f, delimiter='\t')  # CHANGED: was delimiter=','
            writer.writerow([
                'product_id', 'product_name',
                'weekly_revenue', 'units_sold', 'avg_order_value'
            ])
            for row in rows:
                writer.writerow([
                    row['product_id'], row['product_name'],
                    row['weekly_revenue'], row['units_sold'],
                    row['avg_order_value']
                ])
    # Output (TSV): product_id\tproduct_name\tweekly_revenue\tunits_sold\tavg_order_value
    # Example row: P001\tWidget, Deluxe Edition\t4892.50\t217\t22.55

The downstream task ComputeSalesMetrics was not modified. It still called pd.read_csv(self.input().path) with no sep argument, meaning pandas used its default comma delimiter to read what was now a tab-separated file. The result was a DataFrame with a single column — a column whose name was the literal string 'product_id\tproduct_name\tweekly_revenue\tunits_sold\tavg_order_value' — and whose rows were the full tab-separated data lines. The numeric columns weekly_revenue, units_sold, and avg_order_value did not exist as separate DataFrame columns. The task used df.get('weekly_revenue', pd.Series([0])), which silently returned the default when the column was absent:

class ComputeSalesMetrics(luigi.Task):
    week_start = luigi.DateParameter()

    def requires(self):
        return ExtractSalesData(week_start=self.week_start)

    def output(self):
        return luigi.contrib.s3.S3Target(
            f's3://pinecrest-data/reports/metrics-{self.week_start}.json'
        )

    def run(self):
        df = pd.read_csv(self.input().path)
        # BUG: pd.read_csv default sep=',' reads TSV as one-column DataFrame
        #
        # What pandas actually sees:
        #   df.columns = Index(['product_id\tproduct_name\tweekly_revenue\tunits_sold\tavg_order_value'])
        #   df.shape = (N, 1)  — N rows, 1 column
        #   df['weekly_revenue']  → KeyError (column does not exist)
        #
        # df.get() returns the default when the key is absent — no exception raised:
        #   df.get('weekly_revenue', pd.Series([0]))  → pd.Series([0])
        #   pd.Series([0]).sum()                      → 0
        #   float(0)                                  → 0.0

        metrics = {
            'total_revenue': float(df.get('weekly_revenue', pd.Series([0])).sum()),
            'total_units':   float(df.get('units_sold',     pd.Series([0])).sum()),
            'avg_order_value': float(df.get('avg_order_value', pd.Series([0])).mean()),
        }
        with self.output().open('w') as f:
            json.dump(metrics, f)
        # Output written to S3:
        # {"total_revenue": 0.0, "total_units": 0.0, "avg_order_value": 0.0}
        #
        # Luigi checks: does s3://pinecrest-data/reports/metrics-2026-08-18.json exist? Yes.
        # Luigi marks ComputeSalesMetrics complete(). Pipeline: SUCCESS.
        # No exception. No warning. No alert.

With the metrics JSON written and the S3Target confirmed as present, Luigi marked both tasks as complete() and reported the full WeeklySalesReportPipeline as SUCCESS. The AWS Glue downstream job consumed the metrics JSON, found valid JSON with syntactically correct numeric values (0.0 is a valid float), and generated a PDF report that was properly formatted with $0.00 in every revenue cell. The PDF was emailed to all six regional sales managers every Monday morning from August 18 through September 29 — seven reports, all showing $0.

Why seven weeks passed before anyone escalated

The six regional sales managers who received the Monday reports noticed the $0 figures but did not immediately escalate. The first week (August 18–24) was a slow period following a company-wide summer close, and several managers assumed the report reflected reduced activity. By the second week (August 25–31), managers were checking CRM records manually and saw sales activity that contradicted the report, but the discrepancy was attributed to a known reporting lag — the assumption being that the automated system was “in maintenance” or that the pipeline had a temporary data availability issue. The third and fourth weeks continued the same pattern: managers relied on CRM data for their own regional tracking and deferred to the automated report as a summary that would “catch up when the issue was resolved.” No manager escalated because each individually assumed the issue was already known to the data team and was being handled. The data team, meanwhile, was observing a pipeline that ran successfully every Monday and reported no errors — because, from Luigi’s perspective, the pipeline was completing successfully. Luigi Central Scheduler showed all tasks green. There were no Airflow-style alerts on task failure because no task had failed. The pipeline was silently producing incorrect output while appearing entirely healthy.

The failure mode is structurally distinct from a broken pipeline. A broken pipeline fails visibly: tasks error, the scheduler marks them failed, alerts fire, and engineers investigate. A pipeline that silently produces wrong output is harder to detect because every observable signal — task completion, scheduler status, output file existence, downstream job execution — indicates success. The only signal that something was wrong was the content of the output, and that content was only reviewed by humans reading the final PDF report, not by any automated validator. The 49-day gap between the change and the escalation reflects how long a silent content failure can persist when the pipeline infrastructure layer has no mechanism to validate that output values satisfy business-logic constraints.

Root cause diagnosis: Luigi’s complete() model does not validate output content

Luigi’s task completion model is intentionally minimal: a task is considered complete if its output() target exists. For S3Target, this means the S3 object exists at the specified key. This design is correct and appropriate for Luigi’s use case — idempotent pipelines where re-running an already-complete task would be wasteful. But it means that a task whose run() method produces a structurally valid but semantically incorrect output file will be marked complete() and will not be re-run, even if the output values are wrong. The fix for the Pinecrest Analytics Partners case required three changes:

# Fix 1: Update pd.read_csv to use sep='\t' in ComputeSalesMetrics
# pipelines/sales_reporting.py

class ComputeSalesMetrics(luigi.Task):
    week_start = luigi.DateParameter()

    def requires(self):
        return ValidateSalesDataSchema(week_start=self.week_start)
        # FIX: now depends on validation task, not directly on ExtractSalesData

    def output(self):
        return luigi.contrib.s3.S3Target(
            f's3://pinecrest-data/reports/metrics-{self.week_start}.json'
        )

    def run(self):
        df = pd.read_csv(self.input().path, sep='\t')
        # FIX: sep='\t' matches the TSV format written by ExtractSalesData
        # df.columns = Index(['product_id', 'product_name', 'weekly_revenue',
        #                     'units_sold', 'avg_order_value'])
        # df.shape = (N, 5) — correct

        # No longer using .get() with default — KeyError is now an intentional signal
        metrics = {
            'total_revenue':   float(df['weekly_revenue'].sum()),
            'total_units':     float(df['units_sold'].sum()),
            'avg_order_value': float(df['avg_order_value'].mean()),
        }
        with self.output().open('w') as f:
            json.dump(metrics, f)


# Fix 2: Add ValidateSalesDataSchema task between extraction and transformation
class ValidateSalesDataSchema(luigi.Task):
    """
    Reads the first row of ExtractSalesData output and verifies expected columns.
    If the schema does not match, raises ValueError — Luigi marks this task FAILED,
    which causes ComputeSalesMetrics to not run and the pipeline to report FAILED.
    A failed pipeline fires alerts. A pipeline with wrong content does not.
    """
    week_start = luigi.DateParameter()
    EXPECTED_COLUMNS = [
        'product_id', 'product_name',
        'weekly_revenue', 'units_sold', 'avg_order_value'
    ]

    def requires(self):
        return ExtractSalesData(week_start=self.week_start)

    def output(self):
        return luigi.contrib.s3.S3Target(
            f's3://pinecrest-data/sales/validated-{self.week_start}.flag'
        )

    def run(self):
        df_header = pd.read_csv(self.input().path, sep='\t', nrows=0)
        actual_columns = list(df_header.columns)
        if actual_columns != self.EXPECTED_COLUMNS:
            raise ValueError(
                f'Schema mismatch in ExtractSalesData output for {self.week_start}. '
                f'Expected columns: {self.EXPECTED_COLUMNS}. '
                f'Got: {actual_columns}. '
                f'Check delimiter (sep parameter) and column order in ExtractSalesData.run().'
            )
        # Schema validated: write flag file so this task is not re-run
        with self.output().open('w') as f:
            f.write(f'validated:{self.week_start}:columns:{",".join(actual_columns)}\n')
        # If ExtractSalesData is changed to write CSV again, this task raises ValueError
        # → Luigi marks ValidateSalesDataSchema FAILED
        # → ComputeSalesMetrics does not run (dependency not satisfied)
        # → Pipeline reports FAILED → alert fires → engineer investigates

The third change was the historical fix: re-running the pipeline for all seven affected weeks to regenerate the metrics JSON and downstream PDF reports with correct values. Because Luigi marks tasks as complete() based on target existence, the fix required deleting the seven incorrect S3 output files before re-running the pipeline, so Luigi would re-execute the tasks rather than treating them as already complete:

# Historical fix: delete incorrect S3 output files and re-run for all 7 affected weeks
# Run from the project root with AWS credentials configured

import boto3
from datetime import date, timedelta

s3 = boto3.client('s3')
BUCKET = 'pinecrest-data'

# Affected weeks (Monday dates, August 18 through September 29)
affected_weeks = [
    date(2026, 8, 18),
    date(2026, 8, 25),
    date(2026, 9,  1),
    date(2026, 9,  8),
    date(2026, 9, 15),
    date(2026, 9, 22),
    date(2026, 9, 29),
]

for week_start in affected_weeks:
    # Delete the incorrect metrics JSON (so Luigi will re-run ComputeSalesMetrics)
    metrics_key = f'reports/metrics-{week_start}.json'
    s3.delete_object(Bucket=BUCKET, Key=metrics_key)
    print(f'Deleted s3://{BUCKET}/{metrics_key}')

    # Delete the validation flag file (so Luigi will re-run ValidateSalesDataSchema)
    flag_key = f'sales/validated-{week_start}.flag'
    try:
        s3.delete_object(Bucket=BUCKET, Key=flag_key)
    except Exception:
        pass  # May not exist if this is a new task

# The TSV source files (sales/weekly-YYYY-MM-DD.tsv) are correct —
# ExtractSalesData ran correctly, the data is fine, only the downstream
# reading was broken. No need to delete or re-run ExtractSalesData.

# Re-run pipeline for all 7 weeks:
# luigi --module pipelines.sales_reporting WeeklySalesReportPipeline \
#   --week-start 2026-08-18 --local-scheduler
# Repeat for each affected week_start date.
# Each run: ValidateSalesDataSchema reads TSV header, validates columns,
#           writes flag; ComputeSalesMetrics reads TSV with sep='\t',
#           computes correct metrics, writes JSON with correct values.
# AWS Glue downstream job regenerates PDF with correct revenue figures.

Why the defensive df.get() pattern masks schema failures in Luigi pipelines

The df.get('column_name', default_value) pattern is common in pandas code because it avoids KeyError when a column may legitimately be absent in some inputs. In a Luigi pipeline, this pattern has a specific failure mode: when a column is absent due to a schema mismatch (rather than a legitimate absence), df.get() returns the default value silently, the task completes, the S3Target is written, and Luigi marks the task as complete(). The correct pattern for a Luigi task that must have a specific set of columns is to access them directly with df['column_name'], which raises KeyError if the column is absent. A KeyError in Luigi’s run() method causes the task to fail, Luigi marks it as failed, and the pipeline reports FAILED rather than SUCCESS. A failed pipeline fires alerts. A pipeline that produces wrong values does not. The distinction between df.get() and df['column_name'] is the difference between a silent data quality failure and an observable infrastructure failure.

The Pinecrest Analytics Partners fix made this pattern explicit: after adding ValidateSalesDataSchema to verify columns and updating ComputeSalesMetrics to use direct column access rather than df.get(), any future format mismatch between ExtractSalesData and ComputeSalesMetrics will raise an exception, cause a task failure, and surface as a visible pipeline error within minutes of the Monday morning run. The seven-week gap between failure and discovery becomes structurally impossible under the updated pipeline design.

How HourTab helps Luigi developer retainers

The Pinecrest Analytics Partners engagement totaled 3.5 hours: diagnosis (1.0h — 0.25h reviewing Luigi Central Scheduler logs for the week of October 6, confirming all pipeline tasks showed SUCCESS for all seven affected weeks; 0.25h pulling the seven metrics JSON files from S3 and confirming all showed {"total_revenue": 0.0, "total_units": 0.0, "avg_order_value": 0.0}; 0.25h reading the last 50 lines of ComputeSalesMetrics.run() in pipelines/sales_reporting.py and confirming the pd.read_csv call had no sep argument; 0.25h pulling the S3 source TSV file for week of August 18 and confirming it was tab-separated, not comma-separated), fix and deployment (2.0h — 0.25h adding sep='\t' to pd.read_csv in ComputeSalesMetrics; replacing df.get() calls with direct df[column] access; 0.5h writing ValidateSalesDataSchema task; updating ComputeSalesMetrics.requires() to depend on ValidateSalesDataSchema instead of ExtractSalesData directly; 0.25h code review and deployment to production; 0.25h writing the S3 cleanup script for the seven affected weeks; deleting the seven incorrect metrics JSON files; 0.75h re-running pipeline for all seven affected weeks, monitoring each run in Luigi Central Scheduler, confirming correct metric values in regenerated JSON files), post-incident work (0.5h — 0.25h documenting the format contract between ExtractSalesData and downstream tasks in code comments; adding a checklist item to the team’s code review template requiring that any change to a Luigi task’s output format must include simultaneous updates to all tasks that consume that output; 0.25h confirming with the data team that the regenerated PDF reports would be re-emailed to all six regional sales managers for the seven affected weeks).

In a HourTab retainer log, the incident session reads: “WeeklySalesReportPipeline silent failure: Luigi 3.5.1 Python 3.11; ExtractSalesData delimiter changed CSV to TSV August 18 2026 (csv.writer delimiter='\t', output renamed .tsv); ComputeSalesMetrics not updated (pd.read_csv no sep arg, default comma); DataFrame collapses to 1 column; df.get('weekly_revenue') returns pd.Series([0]); metrics JSON all 0.0; S3Target exists Luigi marks SUCCESS; 7 weekly PDFs emailed to 6 region managers August 18 through September 29 all showing $0; VP of Sales escalates October 6 during Q3 board prep; fix: sep='\t' in pd.read_csv; replace df.get() with df[column] for fail-visible KeyError; add ValidateSalesDataSchema task between extraction and transformation; delete 7 incorrect S3 metrics files; re-run pipeline for all 7 affected weeks; 3.5h.”

HourTab turns a time-tracker CSV into a public retainer-hours URL your client can bookmark — no login, no portal, no “how many hours do I have left?” emails. Free for one retainer. $9/mo for up to ten.

FAQ: Luigi developer retainers

See also: Apache Airflow developer retainer · dbt developer retainer · Prefect developer retainer