Add dlt integration for EdX.org S3 data ingestion

[blarghmatey]

Overview

This PR adds dlt (data load tool) integration to a new data_loading code location, enabling Python-based data ingestion as a complement to Airbyte. The initial implementation loads EdX.org database table exports from S3 with proper asset dependency tracking.

What's Added

New Code Location: data_loading

• ✅ Component-aware project structure for clean separation of concerns
• ✅ dlt integration using Dagster's native decorator
• ✅ Environment-based destinations (local/QA/production)
• ✅ Custom asset dependency mapping via
• ✅ Incremental loading based on file modification dates

Dependencies

• dagster-dlt: Native Dagster integration for dlt pipelines
• dlt[filesystem,parquet,pyiceberg]>=1.21.0: Core dlt with Iceberg support
• Dependencies: IAM-based authentication (no secrets in code)

EdX.org S3 Data Source

Implements a production-ready dlt source that:

• Reads TSV files generated by edxorg_archive asset from S3
• Filters to prod sources only (excludes edge/staging)
• Consolidates all courses into non-partitioned Iceberg/Parquet tables
• Tracks one-to-one dependencies on upstream partitioned assets

Source: s3://ol-data-lake-landing-zone-production/edxorg-raw-data/edxorg/raw_data/db_table/{table}/prod/{course}/*.tsv

Destinations:

• Local Dev: file:///.dlt/data (Parquet for fast iteration)
• QA: s3://ol-data-lake-raw-qa/edxorg (Iceberg + Glue: ol_warehouse_qa_raw)
• Production: s3://ol-data-lake-raw-production/edxorg (Iceberg + Glue: ol_warehouse_production_raw)

Asset Architecture

Upstream Assets (edxorg code location)

• Location: dg_projects/edxorg/edxorg/assets/edxorg_db_table_specs.py
• Asset Keys: AssetKey(["edxorg", "raw_data", "db_table", {table}])
• Partitioning: Multi-partitioned by course_id and source_system (prod/edge)
• Type: External AssetSpecs representing TSV files from archive processing

Downstream Assets (data_loading code location)

• Location: dg_projects/data_loading/data_loading/defs/edxorg_s3_ingest/
• Asset Keys: AssetKey(["ol_warehouse_raw_data", "raw__edxorg__s3__tables__{table}"])
• Table Names: raw__edxorg__s3__tables__{table} (matches Airbyte naming pattern)
• Partitioning: None - consolidates all upstream partitions
• Dependencies: One-to-one AssetDep on corresponding upstream table

Dependency Mapping

Custom EdxorgDltTranslator class:

• Overrides get_asset_spec() using modern Dagster API
• Maps each dlt resource to ol_warehouse_raw_data prefix
• Creates non-blocking AssetDep for lineage tracking
• Enables Dagster to show upstream → downstream data flow

Tables Loaded (40+)

User & Authentication:

• auth_user, auth_userprofile, student_anonymoususerid

Enrollment & Progress:

• student_courseenrollment, courseware_studentmodule, student_courseaccessrole

Assessments & Submissions:

• submissions_submission, submissions_score, assessment_* (14 tables)

Certificates & Grading:

• certificates_generatedcertificate, grades_persistentcoursegrade, grades_persistentsubsectiongrade

And 20+ more (teams, wiki, workflow, user profile data)

Key Features

1. Incremental Loading

• Uses dlt's incremental("modification_date") on filesystem source
• Only processes files modified since last successful run
• State persists in .dlt/edxorg_s3/ directory
• Dramatically reduces runtime after initial load

2. Environment-Based Configuration

• Local/CI: Parquet files to local disk (no AWS, no Iceberg overhead)
• QA: Iceberg tables to s3://ol-data-lake-raw-qa/edxorg with Glue catalog
• Production: Iceberg tables to s3://ol-data-lake-raw-production/edxorg with Glue catalog
• Auto-detected via DAGSTER_ENVIRONMENT variable

3. Prod-Only Filtering

• File glob: db_table/{table}/prod/**/*.tsv
• Excludes edge (staging) environment data automatically
• Reduces data volume and processing time

4. IAM-Based Authentication

• Uses AWS IAM roles (IRSA in K8s, ~/.aws/credentials locally)
• No secrets in configuration files
• Credentials resolved at runtime

5. Iceberg + Glue Catalog (QA/Production)

• Native Iceberg support via dlt 1.21.0+ and PyIceberg
• Automatic Glue catalog registration
• Schema evolution support
• Queryable via Athena or Trino

Project Structure

dg_projects/data_loading/
├── .dlt/
│   └── config.toml              # Named destinations (local, qa, production)
├── data_loading/
│   ├── definitions.py           # Code location definitions
│   └── defs/
│       └── edxorg_s3_ingest/
│           ├── dagster_assets.py   # Custom DagsterDltTranslator + @dlt_assets
│           ├── defs.py             # Exports for component discovery
│           ├── loads.py            # dlt source and pipeline
│           └── README.md           # Comprehensive documentation

dg_projects/edxorg/
└── edxorg/
    └── assets/
        └── edxorg_db_table_specs.py  # External AssetSpecs for upstream tables

Usage Examples

Materialize via Dagster UI

1. Navigate to Assets → ol_warehouse_raw_data
2. Select table (e.g., raw__edxorg__s3__tables__auth_user)
3. Click "Materialize"
4. View lineage showing dependency on edxorg/raw_data/db_table/auth_user

Local Testing

# Navigate to data_loading
cd dg_projects/data_loading

# Set environment (optional, defaults to dev/local)
export DAGSTER_ENVIRONMENT=dev

# Run dlt pipeline directly
python -m data_loading.defs.edxorg_s3_ingest.loads

# Query local data with DuckDB
duckdb -c "SELECT * FROM '.dlt/data/raw__edxorg__s3__tables__auth_user/*.parquet' LIMIT 10"

Environment Variables

• DAGSTER_ENVIRONMENT: Controls destination (dev/ci/qa/production)
• DLT_PROJECT_DIR: Auto-set for config path resolution

Technical Implementation

Custom DagsterDltTranslator

class EdxorgDltTranslator(DagsterDltTranslator):
    def get_asset_spec(self, data: DltResourceTranslatorData) -> AssetSpec:
        # Map to ol_warehouse_raw_data prefix
        asset_key = AssetKey(["ol_warehouse_raw_data", resource.name])
        
        # Extract table name and create upstream dependency
        if resource.name.startswith("raw__edxorg__s3__tables__"):
            table_name = resource.name.replace("raw__edxorg__s3__tables__", "")
            deps = [AssetDep(AssetKey(["edxorg", "raw_data", "db_table", table_name]))]
        
        return default_spec.replace_attributes(key=asset_key, deps=deps)

Named Destination Configuration

# .dlt/config.toml
[destination.local]
destination_type = "filesystem"
bucket_url = "file:///.dlt/data"

[destination.qa]
destination_type = "filesystem"
bucket_url = "s3://ol-data-lake-raw-qa/edxorg"

[iceberg_catalog.qa]
iceberg_catalog_name = "ol_warehouse_qa_raw"
iceberg_catalog_type = "glue"

Testing

• ✅ Definitions load without errors in both code locations
• ✅ Asset keys match Airbyte pattern (ol_warehouse_raw_data)
• ✅ Dependencies show correctly in Dagster lineage graph
• ✅ Pipeline configuration resolves environment-specific destinations
• ✅ Custom DagsterDltTranslator uses modern API (no deprecation warnings)
• ⏳ End-to-end data loading requires AWS credentials

Benefits

1. Clean Architecture: Separate code location for ingestion logic
2. Proper Dependencies: Dagster tracks lineage from raw exports → consolidated tables
3. Environment Aware: Same code works across local/QA/production with different configs
4. Incremental: Only processes new/modified files on subsequent runs
5. Production Ready: Iceberg format with Glue catalog integration
6. Flexible: Python-based, easy to extend for new sources
7. LLM-Friendly: dlt's design works well with AI-assisted development

Migration Path

This establishes the foundation for migrating other data sources to dlt:

• REST APIs: Use dlt's rest_api_resources
• File-based sources: Reuse filesystem source pattern
• Custom transforms: Python flexibility vs. Airbyte's JSON configs

Files Changed

New Code Location:

• dg_projects/data_loading/pyproject.toml - Dependencies and metadata
• dg_projects/data_loading/.dlt/config.toml - dlt configuration
• dg_projects/data_loading/data_loading/definitions.py - Code location setup
• dg_projects/data_loading/data_loading/defs/edxorg_s3_ingest/ - Source implementation

EdX.org Updates:

• dg_projects/edxorg/edxorg/assets/edxorg_db_table_specs.py - External asset specs
• dg_projects/edxorg/edxorg/definitions.py - Include db_table specs

References

• dlt Documentation
• dagster-dlt Integration
• DagsterDltTranslator API
• dlt Filesystem Source
• PyIceberg

[Copilot]

Pull request overview

This PR integrates dlt (data load tool) into the lakehouse code location to enable Python-based data ingestion. The initial implementation provides an EdX.org S3 data source that loads TSV files containing database table exports from S3.

Changes:

• Added dlt integration dependencies (dagster-dlt, dlt, duckdb, pyiceberg)
• Implemented EdX.org S3 ingestion pipeline for loading 40+ database tables from S3
• Added SKIP_AIRBYTE environment variable for faster local development iteration
• Created documentation and helper scripts for dlt pipeline development and testing

Reviewed changes

Copilot reviewed 12 out of 14 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
dg_projects/lakehouse/pyproject.toml	Added dlt-related dependencies to project configuration
dg_projects/lakehouse/uv.lock	Lock file updates for new dependencies
dg_projects/lakehouse/lakehouse/definitions.py	Added SKIP_AIRBYTE logic to conditionally load Airbyte assets
dg_projects/lakehouse/lakehouse/defs/edxorg_s3_ingestion/loads.py	Core dlt source implementation for EdX.org S3 data ingestion
dg_projects/lakehouse/lakehouse/defs/edxorg_s3_ingestion/defs.yaml	Dagster component configuration for dlt integration
dg_projects/lakehouse/lakehouse/defs/edxorg_s3_ingestion/README.md	Documentation for EdX.org S3 pipeline
dg_projects/lakehouse/.dlt/config.toml	dlt non-sensitive configuration for local and production destinations
dg_projects/lakehouse/.dlt/secrets.toml.template	Template for dlt credentials configuration
dg_projects/lakehouse/scripts/dlt_env.py	Helper script for managing dlt environment switching
dg_projects/lakehouse/scripts/test_edxorg_import.py	Import validation script for EdX.org pipeline
dg_projects/lakehouse/docs/README.md	Documentation hub for lakehouse guides
dg_projects/lakehouse/docs/local_development.md	Local development guide with SKIP_AIRBYTE usage
.gitignore	Added dlt-specific file patterns to gitignore

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 25 out of 33 changed files in this pull request and generated 16 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[blarghmatey]

Supersedes #1598

[Copilot]

Pull request overview

Copilot reviewed 25 out of 33 changed files in this pull request and generated 13 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[rachellougee]

Initial testing of one table student_courseenrollment looks good. Just a comment about table_format issue I ran into in local testing. We need further testing around incremental loading for large tables

[Copilot]

Pull request overview

Copilot reviewed 20 out of 29 changed files in this pull request and generated 12 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 21 out of 31 changed files in this pull request and generated 5 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[sentry]

Bug: The code calls df.collect() on a Polars LazyFrame after it has already been consumed by df.sink_csv(). A LazyFrame cannot be consumed twice.
_{Severity: HIGH}

Suggested Fix

The emptiness check should be performed before the data is written to the file. Move the if df.collect().is_empty(): block to be before the df.sink_csv() call. This ensures the check is performed on the valid, unconsumed LazyFrame.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent.
Verify if this is a real issue. If it is, propose a fix; if not, explain why it's not
valid.

Location: dg_projects/edxorg/edxorg/assets/edxorg_archive.py#L304

Potential issue: In `edxorg_archive.py`, a Polars LazyFrame `df` is created and then
written to a file using `df.sink_csv()`. Immediately after, the code attempts to call
`df.collect().is_empty()` on the same LazyFrame reference. This is invalid because
`sink_csv()` consumes the LazyFrame, and it cannot be executed or collected a second
time. This will cause a runtime error during the archive processing operation,
preventing the data from being properly ingested.

[rachellougee]

Tested loading a small table, and it works