Building a Sleeper Fantasy Football Lakehouse on Databricks Free Edition
· 54 min read
Motivation and Contents
If you play fantasy football on Sleeper and have ever wanted to perform data analysis without orchestrating spreadsheets and navigating the Sleeper API, this is a completely free solution for you. If you found this blog post, I assume you are comfortable writing a bit of code and enjoy exploring data. That said, I don’t assume that you’re familiar with Databricks. The first section walks through deployment steps, and the rest is a step-by-step guide to building the solution yourself, from software installation to data ingestion to building an ETL pipeline to visualizing data in a dashboard.
The complete source code is available on GitHub. The repository is a Declarative Automation Bundle (DAB), so you can deploy the whole solution with a handful of terminal commands once you point it to your Sleeper league ID. The DAB includes the following resources:
- A volume for storing the raw data
- Schemas following a medallion architecture (more to come on what exactly that is)
- A historical backfill job
- A daily extract job that fetches only the current season
- A Lakeflow Declarative Pipeline that builds the tables
- A Databricks notebook that tours every part of the architecture
- An AI/BI dashboard
Deploying the Completed DAB from the Source Code on GitHub
Databricks Free Edition
First, sign up for a Databricks Free Edition account if you don’t have one already (not to be confused with the deprecated Community Edition accounts). No credit card is required, so you can safely deploy everything in this post without worrying about incurring costs.
Required Software and Installation
This post assumes that you’re using Homebrew as your package manager for code snippets, but these dependencies should be available in whatever other package manager you might use. Run the following commands to ensure you have all the right software installed:
install-dependencies.sh
git clone git@github.com:scottenriquez/sleeper-fantasy-football-databricks-lakehouse.git
cd sleeper-fantasy-football-databricks-lakehouse
brew install uv jq
brew tap databricks/tap && brew install databricks
# developed using 1.14.1
databricks --version
Local Authentication
To deploy the DAB, you must first authenticate with Databricks via the local CLI:
authenticate.sh
# obtain the workspace subdomain from your browser after logging into Databricks
WORKSPACE_SUBDOMAIN="dbc-xxxxxxxx-xxxx"
databricks auth login --host "https://$WORKSPACE_SUBDOMAIN.cloud.databricks.com"
databricks current-user me
Creating a Catalog and Deploying via the CLI
In Databricks, SQL warehouses are dedicated compute resources for running SQL statements (and only SQL statements) against your data lakehouse (a portmanteau of lake and warehouse). In a Free Edition account, Databricks creates one for us automatically called Serverless Starter Warehouse. A catalog is the top level of the three-level namespace used by Databricks: catalog.schema.table. A catalog doesn’t inherently require a SQL warehouse, but this deployment will fail without one because the dashboard (shown later) requires it. One limitation is that it’s currently not possible to create a catalog backed by Default Storage (i.e., fully managed object-storage that Databricks provisions as part of account creation) via DABs. Instead, we must configure this before deployment like so:
create-catalog.sh
# Free Edition has exactly one warehouse
# Free Edition has exactly one warehouse
WAREHOUSE_ID=$(databricks warehouses list -o json | jq -r '.[0].id')
# use CATALOG_NAME=sleeper for production
CATALOG_NAME=sleeper_dev
databricks api post /api/2.0/sql/statements --json "$(jq -n \
--arg warehouse_id "$WAREHOUSE_ID" \
--arg statement "CREATE CATALOG IF NOT EXISTS $CATALOG_NAME" \
'{warehouse_id: $warehouse_id, statement: $statement, wait_timeout: "30s"}')"
Finally, validate and deploy the bundle:
validate-and-deploy-bundle.sh
# replace this variable with your Sleeper league's latest ID
# the easiest way to pull this is from the URL in your browser
export BUNDLE_VAR_root_league_id=1313686753952952320
databricks bundle validate
# target flag refers to the environment
# we'll start with development (dev)
# deploy to production (prod) in order to enable daily refreshes
databricks bundle plan --target dev
databricks bundle deploy --target dev
# tear down resources when done
databricks bundle destroy --target dev
Backfilling Historical Data
As noted above, this project uses a medallion architecture. Simply put, our data flows from raw API responses (Bronze) to typed, deduplicated intermediate data (Silver) to curated, business-level tables (Gold). Gold data is ready for end-user consumption (e.g., dashboards).
With the solution deployed, next run the backfill job for your league to populate historical data. This manual backfill just needs to be done once.
backfill.sh
# each league's season has a different ID with a reference to the previous season
# walk the entire league chain backward and lands every season on the volume
databricks bundle run sleeper_backfill --target dev
# refresh the pipeline to build the tables from the landed JSON
databricks bundle run sleeper_lakehouse_etl --target dev
The backfill status can be monitored via the Workspace UI under the Jobs & Pipelines tab.
Once the data is backfilled, you can now explore it via the dashboard as a quick sanity check. Search for Sleeper League Overview to find it.
Lakehouse Tour Notebook
To review the deployed resources and learn more about our medallion architecture, search for the deployed notebook called lakehouse_tour in the Workspace UI. I’ve also included the notebook below in Jupyter format with outputs for easy review without deploying the bundle.
Building the Solution from Scratch
Scaffolding a New Project
Scaffold the DAB using the Databricks CLI:
scaffold-dab.sh
cat > init_config.json <<'JSON'
{
"project_name": "sleeper_lakehouse",
"include_job": "yes",
"include_pipeline": "yes",
"include_python": "yes",
"serverless": "yes",
"default_catalog": "workspace",
"personal_schemas": "yes"
}
JSON
databricks bundle init default-python --config-file init_config.json --output-dir .
# lift core files to root
for item in .gitignore .vscode AGENTS.md CLAUDE.md databricks.yml fixtures \
pyproject.toml resources src tests README.md; do
mv "sleeper_lakehouse/$item" "./$item"
done
rmdir sleeper_lakehouse && rm init_config.json
Configure uv, add a key dependency for later, and set up formatting:
configure-uv-dependencies-and-ruff.sh
uv sync
uv add requests
uv add --dev responses
cat >> pyproject.toml <<'TOML'
builtins = ["spark", "dbutils", "display", "displayHTML", "sc"]
[tool.ruff.format]
quote-style = "single"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "Q"]
[tool.ruff.lint.flake8-quotes]
inline-quotes = "single"
docstring-quotes = "double"
[tool.pytest.ini_options]
testpaths = ["tests"]
TOML
# some errors are expected from the scaffolded code
uv run ruff check . --fix
uv run ruff format .
The scaffolding ships with a set of NYC taxi samples that demonstrate a job, a notebook, and two pipeline transformations. They are useful for confirming that the bundle works end-to-end, but nothing in this project builds on them. Remove them now so that the first deployment contains only Sleeper resources:
remove-scaffolded-samples.sh
rm -f src/sleeper_lakehouse/main.py \
src/sleeper_lakehouse/taxis.py \
src/sample_notebook.ipynb \
resources/sample_job.job.yml \
src/sleeper_lakehouse_etl/explorations/sample_exploration.ipynb \
src/sleeper_lakehouse_etl/transformations/sample_trips_sleeper_lakehouse.py \
src/sleeper_lakehouse_etl/transformations/sample_zones_sleeper_lakehouse.py \
tests/sample_taxis_test.py
# drop the console script that pointed at the deleted main module
sed -i '' '/^main = "sleeper_lakehouse.main:main"$/d' pyproject.toml
Finally, validate and deploy the initial bundle:
validate-and-deploy-bundle.sh
databricks bundle validate
databricks bundle plan --target dev
databricks bundle deploy --target dev
Declaring Unity Catalog Resources
First, configure these as variables in databricks.yml:
databricks.yml
variables:
schema:
description: The schema to use
raw_schema:
description: Raw API responses
default: raw
bronze_schema:
description: Append-only tables loaded directly from the source volume
default: bronze
silver_schema:
description: Cleansed and conformed data
default: silver
gold_schema:
description: Curated, business-level tables
default: gold
source_volume:
description: >-
Unity Catalog-managed volume that receives extracted JSON files
default: sleeper_api
protect_data:
description: >-
Blocks bundle destroy from deleting catalogs, schemas, and volumes
default: false
root_league_id:
description: >-
Sleeper league identifier for the most recent season; each previous season can be chained from the latest
# replace with your ID
default: '1313686753952952320'
sport:
description: Sleeper sport code used in API paths
default: nfl
sleeper_base_url:
description: Base URL for the public read-only Sleeper API
default: https://api.sleeper.app/v1
Next, we create the Unity Catalog resources:
resources/unity_catalog.yml
resources:
schemas:
raw:
catalog_name: ${var.catalog}
name: ${var.raw_schema}
comment: >-
Landing area for unmodified Sleeper API responses
lifecycle:
prevent_destroy: ${var.protect_data}
bronze:
catalog_name: ${var.catalog}
name: ${var.bronze_schema}
comment: >-
Append-only tables loaded directly from the source volume
lifecycle:
prevent_destroy: ${var.protect_data}
silver:
catalog_name: ${var.catalog}
name: ${var.silver_schema}
comment: >-
Cleansed and conformed data
lifecycle:
prevent_destroy: ${var.protect_data}
gold:
catalog_name: ${var.catalog}
name: ${var.gold_schema}
comment: >-
Curated, business-level tables
lifecycle:
prevent_destroy: ${var.protect_data}
volumes:
source:
catalog_name: ${var.catalog}
schema_name: ${resources.schemas.raw.name}
name: ${var.source_volume}
volume_type: MANAGED
comment: >-
Unity Catalog-managed volume that receives extracted JSON files
lifecycle:
prevent_destroy: ${var.protect_data}
Then, update the variable references in the scaffolded pipeline:
resources/sleeper_lakehouse_etl.pipeline.yml
resources:
pipelines:
sleeper_lakehouse_etl:
name: sleeper_lakehouse_etl
catalog: ${var.catalog}
schema: ${var.schema}
schema: ${resources.schemas.bronze.name}
serverless: true
root_path: '../src/sleeper_lakehouse_etl'
configuration:
sleeper.catalog: ${var.catalog}
sleeper.raw_schema: ${resources.schemas.raw.name}
sleeper.bronze_schema: ${resources.schemas.bronze.name}
sleeper.silver_schema: ${resources.schemas.silver.name}
sleeper.gold_schema: ${resources.schemas.gold.name}
sleeper.source_volume_path: /Volumes/${var.catalog}/${resources.schemas.raw.name}/${resources.volumes.source.name}
libraries:
- glob:
include: ../src/sleeper_lakehouse_etl/transformations/**
Extracting from the Sleeper API
The Sleeper API documentation can be found here. There is no API key required. As long as you know your league’s ID, you can fetch its data via the public API. We’ll use a simple client class for HTTP requests with built-in retry logic and decoding:
src/sleeper_lakehouse/sleeper_client.py
"""HTTP access to the public Sleeper API"""
import time
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
import requests
DEFAULT_BASE_URL = 'https://api.sleeper.app/v1'
RETRYABLE_STATUS_CODES = frozenset({429, 500, 502, 503, 504})
DOCUMENTED_CALLS_PER_MINUTE = 1000
DEFAULT_MINIMUM_INTERVAL_SECONDS = 60 / DOCUMENTED_CALLS_PER_MINUTE
JSONDocument = dict[str, Any] | list[Any] | None
class SleeperAPIError(RuntimeError):
"""Raised when a request cannot be completed after exhausting every attempt"""
@dataclass(frozen=True)
class RetryPolicy:
"""Controls how often and how patiently a failed request is retried"""
maximum_attempts: int = 5
initial_backoff_seconds: float = 0.5
backoff_multiplier: float = 2.0
maximum_backoff_seconds: float = 30.0
class SleeperClient:
"""Reads JSON documents from the Sleeper API"""
def __init__(
self,
base_url: str = DEFAULT_BASE_URL,
session: requests.Session | None = None,
retry_policy: RetryPolicy | None = None,
minimum_interval_seconds: float = DEFAULT_MINIMUM_INTERVAL_SECONDS,
timeout_seconds: float = 30.0,
sleep_function: Callable[[float], None] = time.sleep,
clock_function: Callable[[], float] = time.monotonic,
) -> None:
self.base_url: str = base_url.rstrip('/')
self.session: requests.Session = session if session is not None else requests.Session()
self.retry_policy: RetryPolicy = retry_policy if retry_policy is not None else RetryPolicy()
self.minimum_interval_seconds: float = minimum_interval_seconds
self.timeout_seconds: float = timeout_seconds
self.sleep_function: Callable[[float], None] = sleep_function
self.clock_function: Callable[[], float] = clock_function
self.last_request_started_at: float | None = None
def get(self, path: str) -> JSONDocument:
"""Return the decoded document at ``path``, or ``None`` when it does not exist"""
url: str = f'{self.base_url}/{path.lstrip("/")}'
last_failure: Exception | None = None
for attempt_number in range(1, self.retry_policy.maximum_attempts + 1):
self._wait_for_rate_limit()
self.last_request_started_at = self.clock_function()
try:
response: requests.Response = self.session.get(url, timeout=self.timeout_seconds)
except requests.RequestException as request_error:
last_failure = request_error
else:
if response.status_code == 404:
return None
if response.status_code not in RETRYABLE_STATUS_CODES:
return self._decode(response, url)
last_failure = SleeperAPIError(f'{url} returned HTTP {response.status_code}')
if attempt_number < self.retry_policy.maximum_attempts:
self.sleep_function(self._backoff_seconds(attempt_number, response.headers.get('Retry-After')))
continue
if attempt_number < self.retry_policy.maximum_attempts:
self.sleep_function(self._backoff_seconds(attempt_number, None))
raise SleeperAPIError(f'{url} failed after {self.retry_policy.maximum_attempts} attempts') from last_failure
def _decode(self, response: requests.Response, url: str) -> JSONDocument:
if response.status_code >= 400:
raise SleeperAPIError(f'{url} returned HTTP {response.status_code}')
try:
return response.json()
except ValueError as decode_error:
raise SleeperAPIError(f'{url} returned a body that is not valid JSON') from decode_error
def _wait_for_rate_limit(self) -> None:
if self.last_request_started_at is None:
return
elapsed_seconds: float = self.clock_function() - self.last_request_started_at
remaining_seconds: float = self.minimum_interval_seconds - elapsed_seconds
if remaining_seconds > 0:
self.sleep_function(remaining_seconds)
def _backoff_seconds(self, attempt_number: int, retry_after_header: str | None) -> float:
if retry_after_header:
try:
return min(float(retry_after_header), self.retry_policy.maximum_backoff_seconds)
except ValueError:
pass
delay_seconds: float = self.retry_policy.initial_backoff_seconds * (
self.retry_policy.backoff_multiplier ** (attempt_number - 1)
)
return min(delay_seconds, self.retry_policy.maximum_backoff_seconds)
With this client in hand, we can easily build a Python function for each endpoint:
src/sleeper_lakehouse/endpoints.py
"""Path construction for every Sleeper API endpoint"""
DEFAULT_SPORT = 'nfl'
def _clean_path_segment(value: str, field_name: str) -> str:
"""Return ``value`` stripped, rejecting anything that cannot sit in a single path position"""
if not isinstance(value, str):
raise ValueError(f'{field_name} must be a string but received {value!r}')
stripped_value: str = value.strip()
if not stripped_value:
raise ValueError(f'{field_name} must not be empty but received {value!r}')
if '/' in stripped_value:
raise ValueError(f'{field_name} must not contain a slash but received {value!r}')
return stripped_value
def _clean_week(week: int) -> int:
"""Return ``week`` unchanged, rejecting values that cannot be a real week number"""
if isinstance(week, bool) or not isinstance(week, int):
raise ValueError(f'week must be an integer but received {week!r}')
if week < 1:
raise ValueError(f'week must be 1 or greater but received {week}')
return week
def state(sport: str = DEFAULT_SPORT) -> str:
"""Current season, week, and season type for a sport"""
cleaned_sport: str = _clean_path_segment(sport, 'sport')
return f'state/{cleaned_sport}'
def players(sport: str = DEFAULT_SPORT) -> str:
"""Every player in the sport, roughly 15 MB, so fetch at most once per day"""
cleaned_sport: str = _clean_path_segment(sport, 'sport')
return f'players/{cleaned_sport}'
def league(league_id: str) -> str:
"""Settings, scoring rules, status, and the link to the previous season"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
return f'league/{cleaned_league_id}'
def rosters(league_id: str) -> str:
"""One entry per team, holding owner, players, and season totals"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
return f'league/{cleaned_league_id}/rosters'
def users(league_id: str) -> str:
"""One entry per league member"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
return f'league/{cleaned_league_id}/users'
def matchups(league_id: str, week: int) -> str:
"""Points and starters for every team in a single week"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
cleaned_week: int = _clean_week(week)
return f'league/{cleaned_league_id}/matchups/{cleaned_week}'
def transactions(league_id: str, week: int) -> str:
"""Trades, waiver claims, and free agent moves processed in a single week"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
cleaned_week: int = _clean_week(week)
return f'league/{cleaned_league_id}/transactions/{cleaned_week}'
def winners_bracket(league_id: str) -> str:
"""Playoff bracket for teams competing for the championship"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
return f'league/{cleaned_league_id}/winners_bracket'
def losers_bracket(league_id: str) -> str:
"""Consolation bracket for teams eliminated from the championship"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
return f'league/{cleaned_league_id}/losers_bracket'
def traded_picks(league_id: str) -> str:
"""Every draft pick that has changed hands"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
return f'league/{cleaned_league_id}/traded_picks'
def drafts(league_id: str) -> str:
"""Draft metadata for a league season"""
cleaned_league_id: str = _clean_path_segment(league_id, 'league_id')
return f'league/{cleaned_league_id}/drafts'
def draft_picks(draft_id: str) -> str:
"""Every pick made in a draft, in order"""
cleaned_draft_id: str = _clean_path_segment(draft_id, 'draft_id')
return f'draft/{cleaned_draft_id}/picks'
Building on endpoints.py, we can generate a plan for which leagues, weeks, etc. to pull data for. Since each Sleeper season has a different ID for the same league, we need to pass the latest season’s ID as the initial input to traverse the path of previous seasons.
src/sleeper_lakehouse/league_chain_planning.py
"""Discovery of which league seasons and which weeks within them hold data"""
from dataclasses import dataclass
from typing import Any
from sleeper_lakehouse import endpoints
from sleeper_lakehouse.sleeper_client import JSONDocument, SleeperClient
TERMINAL_PREVIOUS_LEAGUE_IDS = frozenset({'0', ''})
ABSENT_DRAFT_IDS = frozenset({'0', ''})
LEAGUE_STATUSES_BEFORE_KICKOFF = frozenset({'pre_draft', 'drafting'})
STATE_SEASON_TYPES_BEFORE_KICKOFF = frozenset({'pre', 'off'})
COMPLETED_LEAGUE_STATUS = 'complete'
DEFAULT_MAXIMUM_SEASONS = 25
class LeagueChainError(RuntimeError):
"""Raised when the chain of previous seasons cannot be walked to its beginning"""
@dataclass(frozen=True)
class LeagueSeason:
"""One season of a league, together with the weeks worth extracting"""
league_id: str
season: str
status: str
playoff_week_start: int
final_playoff_round: int
weeks: tuple[int, ...]
draft_id: str = ''
def traverse_league_chain(
client: SleeperClient,
root_league_id: str,
maximum_seasons: int = DEFAULT_MAXIMUM_SEASONS,
) -> list[dict[str, Any]]:
"""Return league documents from ``root_league_id`` backwards, newest season first, stopping at the
truthy string ``'0'`` that Sleeper uses instead of a JSON ``null`` to mark the start of a chain"""
documents: list[dict[str, Any]] = []
visited_league_ids: set[str] = set()
league_id: str = _require_root_league_id(root_league_id)
while league_id and league_id not in TERMINAL_PREVIOUS_LEAGUE_IDS:
if league_id in visited_league_ids:
raise LeagueChainError(f'league chain returns to {league_id} and would produce an infinite loop')
if len(documents) >= maximum_seasons:
raise LeagueChainError(f'league chain is longer than {maximum_seasons} seasons')
visited_league_ids.add(league_id)
document: Any = client.get(endpoints.league(league_id))
if not isinstance(document, dict):
raise LeagueChainError(_missing_league_message(league_id, documents))
documents.append(document)
previous_league_id: Any = document.get('previous_league_id')
league_id = previous_league_id if isinstance(previous_league_id, str) else ''
return documents
def final_playoff_round(bracket: JSONDocument) -> int:
"""Return the highest round number in a playoff bracket or zero when there is none"""
if not isinstance(bracket, list):
return 0
round_numbers: list[int] = [
match['r'] for match in bracket if isinstance(match, dict) and isinstance(match.get('r'), int)
]
return max(round_numbers) if round_numbers else 0
def discover_weeks(
league_document: dict[str, Any],
playoff_round: int,
current_state: dict[str, Any] | None = None,
) -> tuple[int, ...]:
"""Return the weeks of one season that should hold data running through the last playoff round for a
finished season and capped at the current NFL week for one under way, and empty before kickoff"""
settings: Any = league_document.get('settings')
playoff_week_start: Any = settings.get('playoff_week_start') if isinstance(settings, dict) else None
if not isinstance(playoff_week_start, int) or isinstance(playoff_week_start, bool) or playoff_week_start < 1:
return ()
last_week: int = playoff_week_start + max(playoff_round, 0) - 1
if last_week < 1:
return ()
if league_document.get('status') in LEAGUE_STATUSES_BEFORE_KICKOFF:
return ()
if league_document.get('status') == COMPLETED_LEAGUE_STATUS:
return tuple(range(1, last_week + 1))
if current_state is None or current_state.get('season') != league_document.get('season'):
return tuple(range(1, last_week + 1))
if current_state.get('season_type') in STATE_SEASON_TYPES_BEFORE_KICKOFF:
return ()
current_week: Any = current_state.get('week')
if not isinstance(current_week, int) or isinstance(current_week, bool) or current_week < 1:
return ()
return tuple(range(1, min(current_week, last_week) + 1))
def build_extraction_plan(
client: SleeperClient,
root_league_id: str,
sport: str = endpoints.DEFAULT_SPORT,
maximum_seasons: int = DEFAULT_MAXIMUM_SEASONS,
) -> list[LeagueSeason]:
"""Walk the chain and resolve each season into the weeks worth extracting"""
documents: list[dict[str, Any]] = traverse_league_chain(client, root_league_id, maximum_seasons)
current_state: Any = client.get(endpoints.state(sport))
if not isinstance(current_state, dict):
current_state = None
plan: list[LeagueSeason] = []
for document in documents:
league_id: str = str(document.get('league_id') or '')
bracket: Any = client.get(endpoints.winners_bracket(league_id))
playoff_round: int = final_playoff_round(bracket)
settings: Any = document.get('settings')
playoff_week_start: Any = settings.get('playoff_week_start') if isinstance(settings, dict) else None
plan.append(
LeagueSeason(
league_id=league_id,
season=str(document.get('season') or ''),
status=str(document.get('status') or ''),
playoff_week_start=playoff_week_start if isinstance(playoff_week_start, int) else 0,
final_playoff_round=playoff_round,
weeks=discover_weeks(document, playoff_round, current_state),
draft_id=draft_identifier(document),
)
)
return plan
def draft_identifier(league_document: dict[str, Any]) -> str:
"""Return the draft this season points at"""
draft_id: Any = league_document.get('draft_id')
if not isinstance(draft_id, str):
return ''
stripped_draft_id: str = draft_id.strip()
return '' if stripped_draft_id in ABSENT_DRAFT_IDS else stripped_draft_id
def _require_root_league_id(root_league_id: str) -> str:
if not isinstance(root_league_id, str) or not root_league_id.strip():
raise LeagueChainError(f'root_league_id must be a non-empty string, received {root_league_id!r}')
stripped_root_league_id: str = root_league_id.strip()
if stripped_root_league_id in TERMINAL_PREVIOUS_LEAGUE_IDS:
raise LeagueChainError(
f'root_league_id {stripped_root_league_id!r} is the end-of-chain marker rather than a valid ID'
)
return stripped_root_league_id
def _missing_league_message(league_id: str, documents: list[dict[str, Any]]) -> str:
if not documents:
return f'root league {league_id} does not exist'
previous_season: Any = documents[-1].get('season')
return f'league {league_id} does not exist, but the {previous_season} season names it as its previous season'
Landing Raw Data in Databricks
We’ll start by building a backfill job to load the fantasy football league’s full history. First, create a job resource in the DAB:
resources/sleeper_backfill.job.yml
resources:
jobs:
sleeper_backfill:
name: sleeper_backfill
description: >-
Parses every season in the league chain; run once after the first deployment
parameters:
- name: seasons
default: all
- name: force
default: "false"
tasks:
- task_key: extract
python_wheel_task:
package_name: sleeper_lakehouse
entry_point: extract
parameters:
- "--volume-path"
- /Volumes/${var.catalog}/${resources.schemas.raw.name}/${resources.volumes.source.name}
- "--root-league-id"
- "${var.root_league_id}"
- "--sport"
- "${var.sport}"
- "--seasons"
- "{{job.parameters.seasons}}"
- "--force"
- "{{job.parameters.force}}"
environment_key: default
environments:
- environment_key: default
spec:
environment_version: "4"
dependencies:
- ../dist/*.whl
Next, we need to modify the pyproject.toml file so that the job can find the extract code:
pyproject.toml
dependencies = [
"requests>=2.32",
]
[project.scripts]
extract = "sleeper_lakehouse.extract_job:main"
Adding an extract script lets the entry_point: extract configuration in the job run the Python code. Building the wheel bakes [project.scripts] into its metadata, and python_wheel_task resolves the name from there:
sleeper_lakehouse-0.0.1.dist-info/entry_points.txt
[console_scripts]
extract = sleeper_lakehouse.extract_job:main
Create the entry point called extract_job.py that lands Sleeper API responses on a Unity Catalog volume. This script creates the CLI that the job will use. It creates an ArgumentParser, handles inputs, instantiates the SleeperClient class, and invokes the extraction logic.
src/sleeper_lakehouse/extract_job.py
"""Job entry point that lands Sleeper API responses on a Unity Catalog volume"""
import argparse
from sleeper_lakehouse import endpoints
from sleeper_lakehouse.extraction import (
DEFAULT_PLAYERS_REFRESH_DAYS,
DEFAULT_REVISION_WEEKS,
SEASONS_CURRENT,
ExtractionSummary,
extract,
)
from sleeper_lakehouse.landing_writer import LandingWriter
from sleeper_lakehouse.sleeper_client import SleeperClient
TRUE_VALUES = frozenset({'true', 'yes', '1'})
FALSE_VALUES = frozenset({'false', 'no', '0'})
def parse_boolean(value: str) -> bool:
"""Read a boolean from a job parameter"""
lowered: str = value.strip().lower()
if lowered in TRUE_VALUES:
return True
if lowered in FALSE_VALUES:
return False
raise argparse.ArgumentTypeError(f'Expected one of {sorted(TRUE_VALUES | FALSE_VALUES)} but received {value!r}')
def build_parser() -> argparse.ArgumentParser:
"""Create and configure argument parser"""
parser = argparse.ArgumentParser(description='Land Sleeper API responses on a Unity Catalog volume')
parser.add_argument('--volume-path', required=True, help='Root of the landing volume')
parser.add_argument('--root-league-id', required=True, help='Most recent season of the league')
parser.add_argument('--seasons', default=SEASONS_CURRENT, help='All or current')
parser.add_argument('--sport', default=endpoints.DEFAULT_SPORT)
parser.add_argument('--revision-weeks', type=int, default=DEFAULT_REVISION_WEEKS)
parser.add_argument('--players-refresh-days', type=int, default=DEFAULT_PLAYERS_REFRESH_DAYS)
parser.add_argument('--force', type=parse_boolean, default=False)
return parser
def main() -> None:
"""Run one extraction and fail the task when any response could not be landed"""
arguments = build_parser().parse_args()
print(f'Landing into {arguments.volume_path}')
print(f'seasons={arguments.seasons} sport={arguments.sport} force={arguments.force}')
print(f'revision_weeks={arguments.revision_weeks} players_refresh_days={arguments.players_refresh_days}')
summary: ExtractionSummary = extract(
client=SleeperClient(),
writer=LandingWriter(arguments.volume_path),
root_league_id=arguments.root_league_id,
seasons=arguments.seasons,
sport=arguments.sport,
revision_weeks=arguments.revision_weeks,
force=arguments.force,
players_refresh_days=arguments.players_refresh_days,
)
print(
f'fetched={summary.fetched} skipped={summary.skipped} records={summary.record_count} bytes={summary.byte_count}'
)
for failure in summary.failures:
print(f'Failed: {failure}')
summary.raise_for_failures()
if __name__ == '__main__':
main()
Now add the extraction.py module. The extract method starts execution. It first creates a plan using the league_chain_planning.py logic above, builds a list of ExtractionTasks to execute, and finally runs them.
src/sleeper_lakehouse/extraction.py
"""Core extraction logic"""
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Protocol
from sleeper_lakehouse import endpoints
from sleeper_lakehouse.landing import LandingEnvelope
from sleeper_lakehouse.landing_writer import LandingResult
from sleeper_lakehouse.league_chain_planning import (
COMPLETED_LEAGUE_STATUS,
LeagueSeason,
build_extraction_plan,
)
from sleeper_lakehouse.sleeper_client import DEFAULT_BASE_URL, JSONDocument, SleeperClient
SEASONS_ALL = 'all'
SEASONS_CURRENT = 'current'
SEASON_SCOPES = frozenset({SEASONS_ALL, SEASONS_CURRENT})
REFRESH_ONCE: int | None = None
REFRESH_DAILY: int = 1
REFRESH_WEEKLY: int = 7
DEFAULT_REVISION_WEEKS = 3
DEFAULT_PLAYERS_REFRESH_DAYS = REFRESH_WEEKLY
PER_SEASON_ENTITIES: tuple[tuple[str, Callable[[str], str]], ...] = (
('league', endpoints.league),
('rosters', endpoints.rosters),
('users', endpoints.users),
('winners_bracket', endpoints.winners_bracket),
('losers_bracket', endpoints.losers_bracket),
('traded_picks', endpoints.traded_picks),
('drafts', endpoints.drafts),
)
PER_WEEK_ENTITIES: tuple[tuple[str, Callable[[str, int], str]], ...] = (
('matchups', endpoints.matchups),
('transactions', endpoints.transactions),
)
PER_DRAFT_ENTITIES: tuple[tuple[str, Callable[[str], str]], ...] = (('draft_picks', endpoints.draft_picks),)
class LandingTarget(Protocol):
"""Where landed records are written and is satisfied by both the FUSE and Files API writers"""
def write(
self,
document: JSONDocument,
envelope: LandingEnvelope,
mapping_key_field: str | None = None,
) -> LandingResult: ...
def has_landed(self, envelope: LandingEnvelope) -> bool: ...
def landed_within_days(self, envelope: LandingEnvelope, days: int) -> bool: ...
class ExtractionError(RuntimeError):
"""Raised when one or more responses could not be landed"""
@dataclass(frozen=True)
class ExtractionTask:
"""One response to fetch and land"""
entity: str
path: str
envelope: LandingEnvelope
refresh_days: int | None
mapping_key_field: str | None = None
@dataclass(frozen=True)
class ExtractionSummary:
"""Counts describing what one extraction run did"""
fetched: int = 0
skipped: int = 0
record_count: int = 0
byte_count: int = 0
failures: tuple[str, ...] = field(default_factory=tuple)
def raise_for_failures(self) -> None:
"""Fail the run when any response could not be landed, so a partial run is never reported as success"""
if self.failures:
raise ExtractionError(f'{len(self.failures)} responses failed to land: ' + '; '.join(self.failures))
def select_seasons(plan: list[LeagueSeason], seasons: str) -> list[LeagueSeason]:
"""Narrow a plan to every season or to the most recent one"""
if seasons == SEASONS_ALL:
return list(plan)
if seasons == SEASONS_CURRENT:
return list(plan[:1])
raise ValueError(f'seasons must be one of {sorted(SEASON_SCOPES)} but received {seasons!r}')
def revisable_weeks(season: LeagueSeason, revision_weeks: int) -> frozenset[int]:
"""Return the trailing weeks of an unfinished season that may still change through stat corrections"""
if season.status == COMPLETED_LEAGUE_STATUS or revision_weeks < 1 or not season.weeks:
return frozenset()
return frozenset(season.weeks[-revision_weeks:])
def build_tasks(
plan: list[LeagueSeason],
fetched_at: datetime,
sport: str = endpoints.DEFAULT_SPORT,
revision_weeks: int = DEFAULT_REVISION_WEEKS,
base_url: str = DEFAULT_BASE_URL,
players_refresh_days: int = DEFAULT_PLAYERS_REFRESH_DAYS,
) -> list[ExtractionTask]:
"""Turn a plan into the ordered list of responses to fetch"""
tasks: list[ExtractionTask] = []
for season in plan:
season_refresh: int | None = REFRESH_ONCE if season.status == COMPLETED_LEAGUE_STATUS else REFRESH_DAILY
for entity, build_path in PER_SEASON_ENTITIES:
path: str = build_path(season.league_id)
tasks.append(
ExtractionTask(
entity=entity,
path=path,
envelope=_build_envelope(entity, path, fetched_at, base_url, season.league_id, season.season),
refresh_days=season_refresh,
)
)
if season.draft_id:
for entity, build_draft_path in PER_DRAFT_ENTITIES:
draft_path: str = build_draft_path(season.draft_id)
tasks.append(
ExtractionTask(
entity=entity,
path=draft_path,
envelope=_build_envelope(
entity, draft_path, fetched_at, base_url, season.league_id, season.season
),
refresh_days=season_refresh,
)
)
mutable_weeks: frozenset[int] = revisable_weeks(season, revision_weeks)
for week in season.weeks:
week_refresh: int | None = REFRESH_DAILY if week in mutable_weeks else REFRESH_ONCE
for entity, build_week_path in PER_WEEK_ENTITIES:
week_path: str = build_week_path(season.league_id, week)
tasks.append(
ExtractionTask(
entity=entity,
path=week_path,
envelope=_build_envelope(
entity, week_path, fetched_at, base_url, season.league_id, season.season, week
),
refresh_days=week_refresh,
)
)
tasks.extend(_global_tasks(fetched_at, sport, base_url, players_refresh_days))
return tasks
def run_tasks(
client: SleeperClient,
writer: LandingTarget,
tasks: list[ExtractionTask],
force: bool = False,
) -> ExtractionSummary:
"""Fetch and land every task that is not already satisfied"""
fetched: int = 0
skipped: int = 0
record_count: int = 0
byte_count: int = 0
failures: list[str] = []
for task in tasks:
if not force and _already_satisfied(writer, task):
skipped += 1
continue
try:
document = client.get(task.path)
result = writer.write(document, task.envelope, task.mapping_key_field)
except Exception as error:
failures.append(f'{task.path} ({type(error).__name__}: {error})')
continue
fetched += 1
record_count += result.record_count
byte_count += result.byte_count
return ExtractionSummary(
fetched=fetched,
skipped=skipped,
record_count=record_count,
byte_count=byte_count,
failures=tuple(failures),
)
def extract(
client: SleeperClient,
writer: LandingTarget,
root_league_id: str,
seasons: str = SEASONS_CURRENT,
sport: str = endpoints.DEFAULT_SPORT,
revision_weeks: int = DEFAULT_REVISION_WEEKS,
force: bool = False,
fetched_at: datetime | None = None,
players_refresh_days: int = DEFAULT_PLAYERS_REFRESH_DAYS,
) -> ExtractionSummary:
"""Discover the league chain, decide what should be fetched, and land it"""
moment: datetime = fetched_at if fetched_at is not None else datetime.now(timezone.utc)
plan: list[LeagueSeason] = build_extraction_plan(client, root_league_id, sport)
selected: list[LeagueSeason] = select_seasons(plan, seasons)
tasks: list[ExtractionTask] = build_tasks(
selected, moment, sport, revision_weeks, client.base_url, players_refresh_days
)
return run_tasks(client, writer, tasks, force)
def _global_tasks(
fetched_at: datetime,
sport: str,
base_url: str,
players_refresh_days: int,
) -> list[ExtractionTask]:
state_path: str = endpoints.state(sport)
players_path: str = endpoints.players(sport)
return [
ExtractionTask(
entity='state',
path=state_path,
envelope=_build_envelope('state', state_path, fetched_at, base_url),
refresh_days=REFRESH_DAILY,
),
ExtractionTask(
entity='players',
path=players_path,
envelope=_build_envelope('players', players_path, fetched_at, base_url),
refresh_days=players_refresh_days,
mapping_key_field='player_id',
),
]
def _build_envelope(
entity: str,
path: str,
fetched_at: datetime,
base_url: str,
league_id: str | None = None,
season: str | None = None,
week: int | None = None,
) -> LandingEnvelope:
return LandingEnvelope(
entity=entity,
source_url=f'{base_url.rstrip("/")}/{path}',
fetched_at=fetched_at,
league_id=league_id,
season=season,
week=week,
)
def _already_satisfied(writer: LandingTarget, task: ExtractionTask) -> bool:
if task.refresh_days is None:
return writer.has_landed(task.envelope)
return writer.landed_within_days(task.envelope, task.refresh_days)
Finally, add the landing.py and landing_writer.py files that provide write functionality for the ExtractionTasks:
src/sleeper_lakehouse/landing.py
"""Framing and volume paths for JSON landed from the Sleeper API"""
import json
from dataclasses import dataclass
from datetime import date, datetime, timezone
from pathlib import PurePosixPath
from typing import Any
from sleeper_lakehouse.sleeper_client import JSONDocument
ENTITIES = frozenset(
{
'league',
'rosters',
'users',
'matchups',
'transactions',
'winners_bracket',
'losers_bracket',
'traded_picks',
'drafts',
'draft_picks',
'state',
'players',
}
)
LANDING_SUFFIX = '.jsonl'
UNNAMED_VALUE_FIELD = '_value'
FETCHED_PREFIX = 'fetched='
@dataclass(frozen=True)
class LandingEnvelope:
"""Provenance stamped onto every record landed from one API response"""
entity: str
source_url: str
fetched_at: datetime
league_id: str | None = None
season: str | None = None
week: int | None = None
def envelope_fields(envelope: LandingEnvelope) -> dict[str, Any]:
"""Return the underscore-prefixed provenance keys merged into every landed record"""
fields: dict[str, Any] = {
'_entity': _clean_entity(envelope.entity),
'_source_url': envelope.source_url,
'_fetched_at': format_timestamp(envelope.fetched_at),
}
if envelope.league_id is not None:
fields['_league_id'] = _clean_segment(envelope.league_id, 'league_id')
if envelope.season is not None:
fields['_season'] = _clean_segment(envelope.season, 'season')
if envelope.week is not None:
fields['_week'] = _clean_week(envelope.week)
return fields
def frame_records(
document: JSONDocument,
envelope: LandingEnvelope,
mapping_key_field: str | None = None,
) -> list[dict[str, Any]]:
"""Return one record per entity in ``document`` with provenance merged in, or nothing when it is absent"""
provenance: dict[str, Any] = envelope_fields(envelope)
if document is None:
return []
if isinstance(document, list):
return [{**_as_record(element), **provenance} for element in document]
if mapping_key_field is not None:
return [{**_as_record(value), mapping_key_field: key, **provenance} for key, value in sorted(document.items())]
return [{**document, **provenance}]
def serialize_records(records: list[dict[str, Any]]) -> str:
"""Return newline-delimited JSON sorted by key"""
return ''.join(json.dumps(record, ensure_ascii=False, sort_keys=True) + '\n' for record in records)
def landing_path(envelope: LandingEnvelope) -> PurePosixPath:
"""Return the volume-relative path holding this response"""
fetch_date: str = envelope.fetched_at.astimezone(timezone.utc).date().isoformat()
return _landing_directory(envelope, f'{FETCHED_PREFIX}{fetch_date}') / _landing_filename(envelope)
def landing_glob(envelope: LandingEnvelope) -> str:
"""Return a pattern matching this response under any fetch date"""
return str(_landing_directory(envelope, f'{FETCHED_PREFIX}*') / _landing_filename(envelope))
def fetch_date_from_path(path: PurePosixPath | str) -> date | None:
"""Return the fetch date encoded in a landed path"""
for component in PurePosixPath(path).parts:
if not component.startswith(FETCHED_PREFIX):
continue
try:
return date.fromisoformat(component[len(FETCHED_PREFIX) :])
except ValueError:
return None
return None
def _landing_directory(envelope: LandingEnvelope, fetched_component: str) -> PurePosixPath:
directory: PurePosixPath = PurePosixPath(_clean_entity(envelope.entity))
if envelope.season is not None:
directory = directory / f'season={_clean_segment(envelope.season, "season")}'
return directory / fetched_component
def format_timestamp(moment: datetime) -> str:
"""Return an ISO 8601 timestamp in UTC"""
if not isinstance(moment, datetime):
raise ValueError(f'fetched_at must be a datetime but received {moment!r}')
if moment.tzinfo is None:
raise ValueError('fetched_at must be timezone-aware so the landed timestamp is unambiguous')
return moment.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
def _landing_filename(envelope: LandingEnvelope) -> str:
name_parts: list[str] = []
if envelope.league_id is not None:
name_parts.append(f'league={_clean_segment(envelope.league_id, "league_id")}')
if envelope.week is not None:
name_parts.append(f'week={_clean_week(envelope.week):02d}')
if not name_parts:
name_parts.append(_clean_entity(envelope.entity))
return '_'.join(name_parts) + LANDING_SUFFIX
def _as_record(element: object) -> dict[str, Any]:
return element if isinstance(element, dict) else {UNNAMED_VALUE_FIELD: element}
def _clean_entity(entity: str) -> str:
cleaned_entity: str = _clean_segment(entity, 'entity')
if cleaned_entity not in ENTITIES:
raise ValueError(f'entity must be one of {sorted(ENTITIES)} but received {entity!r}')
return cleaned_entity
def _clean_segment(value: str, field_name: str) -> str:
if not isinstance(value, str):
raise ValueError(f'{field_name} must be a string but received {value!r}')
stripped_value: str = value.strip()
if not stripped_value:
raise ValueError(f'{field_name} must not be empty but received {value!r}')
if '/' in stripped_value or '\\' in stripped_value:
raise ValueError(f'{field_name} must not contain a path separator but received {value!r}')
if stripped_value in {'.', '..'}:
raise ValueError(f'{field_name} must not be a relative path marker but received {value!r}')
return stripped_value
def _clean_week(week: int) -> int:
if isinstance(week, bool) or not isinstance(week, int):
raise ValueError(f'Week must be an integer but received {week!r}')
if week < 1:
raise ValueError(f'Week must be 1 or greater but received {week}')
return week
src/sleeper_lakehouse/landing_writer.py
"""Writing landed records beneath a Unity Catalog volume or any other POSIX path"""
from dataclasses import dataclass
from datetime import date, timedelta, timezone
from pathlib import Path, PurePosixPath
from sleeper_lakehouse.landing import (
LandingEnvelope,
fetch_date_from_path,
frame_records,
landing_glob,
landing_path,
serialize_records,
)
from sleeper_lakehouse.sleeper_client import JSONDocument
ENCODING = 'utf-8'
@dataclass(frozen=True)
class LandingResult:
"""Result of landing one API response"""
relative_path: PurePosixPath
record_count: int
byte_count: int
class LandingWriter:
"""Writes newline-delimited JSON under a root directory"""
def __init__(self, root: str | Path) -> None:
self.root: Path = _clean_root(root)
def write(
self,
document: JSONDocument,
envelope: LandingEnvelope,
mapping_key_field: str | None = None,
) -> LandingResult:
"""Land one response and return what was written"""
records: list[dict[str, object]] = frame_records(document, envelope, mapping_key_field)
text: str = serialize_records(records)
relative_path: PurePosixPath = landing_path(envelope)
destination: Path = self.root / relative_path
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(text, encoding=ENCODING)
return LandingResult(
relative_path=relative_path,
record_count=len(records),
byte_count=len(text.encode(ENCODING)),
)
def has_landed(self, envelope: LandingEnvelope) -> bool:
"""Report whether this response was landed by any earlier run"""
return any(self.root.glob(landing_glob(envelope)))
def has_landed_for_fetch_date(self, envelope: LandingEnvelope) -> bool:
"""Report whether this response was already landed under this envelope's own fetch date"""
return self.absolute_path(envelope).exists()
def landed_within_days(self, envelope: LandingEnvelope, days: int) -> bool:
"""Report whether this response was landed recently enough to still count as fresh"""
if days < 1:
return False
if days == 1:
return self.has_landed_for_fetch_date(envelope)
cutoff: date = envelope.fetched_at.astimezone(timezone.utc).date() - timedelta(days=days - 1)
for candidate in self.root.glob(landing_glob(envelope)):
landed_on: date | None = fetch_date_from_path(candidate.relative_to(self.root))
if landed_on is not None and landed_on >= cutoff:
return True
return False
def absolute_path(self, envelope: LandingEnvelope) -> Path:
"""Return where this response would be written under the configured root"""
return self.root / landing_path(envelope)
def _clean_root(root: str | Path) -> Path:
if isinstance(root, Path):
return root
if not isinstance(root, str):
raise ValueError(f'Root must be a string or Path but received {root!r}')
stripped_root: str = root.strip()
if not stripped_root:
raise ValueError(f'Root must not be empty but received {root!r}')
return Path(stripped_root)
Building a Medallion Architecture Pipeline
Everything up to this point lands JSON Lines on a volume. Turning that into tables is the job of a single Lakeflow Declarative Pipeline. The Free Edition allows only one active pipeline per pipeline type, so all three layers live in one pipeline rather than one pipeline per layer. Before the transformations, we need to add one configuration utility to accommodate our code structure. More detail can be found here.
src/sleeper_lakehouse_etl/utilities/project_path.py
"""Making the project package importable from inside a pipeline"""
import sys
from pyspark.sql import SparkSession
SOURCE_ROOT_KEY = 'sleeper.source_root'
def pipeline_configuration(key: str) -> str:
"""Read one required value from the pipeline configuration"""
session: SparkSession | None = SparkSession.getActiveSession()
if session is None:
raise RuntimeError('A pipeline configuration can only be read from an active Spark session')
configured_value: str | None = session.conf.get(key, None)
if configured_value is None:
raise ValueError(f'{key} must be set in the pipeline configuration')
return configured_value
def add_project_package_to_path() -> None:
"""Put the deployed source root on the path because it sits outside the pipeline root folder"""
source_root: str = pipeline_configuration(SOURCE_ROOT_KEY)
if source_root not in sys.path:
sys.path.insert(0, source_root)
Bronze
Recall from the notebook tour above that Bronze mirrors the API and applies no business logic. Every entity becomes one streaming table loaded by Auto Loader. Let’s start by defining our Bronze configurations:
src/sleeper_lakehouse/bronze_config.py
"""Table names, source paths, and reader options for the Bronze layer"""
from dataclasses import dataclass, field
from sleeper_lakehouse.landing import ENTITIES
RESCUED_DATA_COLUMN = '_rescued_data'
AUTO_LOADER_FORMAT = 'json'
SCHEMA_HINTS_OPTION = 'cloudFiles.schemaHints'
AUTO_LOADER_BASE_OPTIONS: dict[str, str] = {
'cloudFiles.format': AUTO_LOADER_FORMAT,
'cloudFiles.inferColumnTypes': 'true',
'cloudFiles.schemaEvolutionMode': 'addNewColumns',
'cloudFiles.partitionColumns': '',
'rescuedDataColumn': RESCUED_DATA_COLUMN,
}
BRONZE_SCHEMA_HINTS: dict[str, str] = {
'matchups': 'players_points map<string,double>',
'transactions': 'adds map<string,bigint>, drops map<string,bigint>',
'rosters': 'metadata map<string,string>',
'drafts': 'draft_order map<string,bigint>',
'draft_picks': 'metadata map<string,string>',
'players': 'metadata map<string,string>',
'users': 'metadata map<string,string>',
}
BRONZE_ENTITIES: tuple[str, ...] = (
'league',
'rosters',
'users',
'winners_bracket',
'losers_bracket',
'traded_picks',
'drafts',
'draft_picks',
'matchups',
'transactions',
'state',
'players',
)
BRONZE_COMMENTS: dict[str, str] = {
'league': 'Settings, scoring rules, and roster positions for one season of the league',
'rosters': 'One row per team per season holding owner, settings, and player identifiers',
'users': 'Display names and avatars for every manager in a season',
'winners_bracket': 'Playoff matchups for teams competing for the championship',
'losers_bracket': 'Consolation matchups for teams eliminated from the championship',
'traded_picks': 'Draft picks that changed hands, including future seasons',
'drafts': 'Draft metadata for each season, including type, status, and settings',
'draft_picks': 'One row per pick in a draft with the auction amount',
'matchups': 'One row per team per week holding points scored and the lineup that produced them',
'transactions': 'Waivers, free agent moves, and trades recorded for one week',
'state': 'Current NFL season and week as reported by Sleeper',
'players': 'The full NFL player dictionary keyed by player identifier',
}
@dataclass(frozen=True)
class BronzeTable:
"""One Bronze streaming table fed by every landed file for a single entity"""
entity: str
table_name: str
source_path: str
comment: str
reader_options: dict[str, str] = field(default_factory=dict)
def clean_volume_path(volume_path: str) -> str:
"""Return the landing volume root without a trailing separator"""
if not isinstance(volume_path, str):
raise ValueError(f'volume_path must be a string but received {volume_path!r}')
stripped_path: str = volume_path.strip().rstrip('/')
if not stripped_path:
raise ValueError(f'volume_path must not be empty but received {volume_path!r}')
if not stripped_path.startswith('/'):
raise ValueError(f'volume_path must be an absolute volume path but received {volume_path!r}')
return stripped_path
def bronze_source_path(volume_path: str, entity: str) -> str:
"""Return the directory holding every landed file for one entity"""
if entity not in ENTITIES:
raise ValueError(f'entity must be one of {sorted(ENTITIES)} but received {entity!r}')
return f'{clean_volume_path(volume_path)}/{entity}'
def auto_loader_options(entity: str) -> dict[str, str]:
"""Return the reader options for one entity, hinting the columns Sleeper returns as keyed maps"""
options: dict[str, str] = dict(AUTO_LOADER_BASE_OPTIONS)
schema_hints: str | None = BRONZE_SCHEMA_HINTS.get(entity)
if schema_hints is not None:
options[SCHEMA_HINTS_OPTION] = schema_hints
return options
def build_bronze_tables(volume_path: str) -> tuple[BronzeTable, ...]:
"""Return one table definition for every entity the extract job lands"""
return tuple(
BronzeTable(
entity=entity,
table_name=entity,
source_path=bronze_source_path(volume_path, entity),
comment=BRONZE_COMMENTS[entity],
reader_options=auto_loader_options(entity),
)
for entity in BRONZE_ENTITIES
)
src/sleeper_lakehouse_etl/transformations/bronze_sleeper_entities.py
"""Bronze streaming tables fed by the JSON landed on the source volume"""
from utilities.project_path import add_project_package_to_path, pipeline_configuration
add_project_package_to_path()
from pyspark import pipelines as dp
from pyspark.sql import DataFrame
from sleeper_lakehouse.bronze_config import BronzeTable, build_bronze_tables
SOURCE_VOLUME_PATH_KEY = 'sleeper.source_volume_path'
def read_landed_entity(source_path: str, reader_options: dict[str, str]) -> DataFrame:
"""Stream every landed file for one entity, skipping files consumed by an earlier run"""
return spark.readStream.format('cloudFiles').options(**reader_options).load(source_path)
def define_bronze_table(table: BronzeTable) -> None:
"""Register one streaming table, taking the definition as an argument so each closure binds its own entity"""
def load_entity() -> DataFrame:
return read_landed_entity(table.source_path, table.reader_options)
load_entity.__name__ = table.table_name
dp.table(name=table.table_name, comment=table.comment)(load_entity)
for bronze_table in build_bronze_tables(pipeline_configuration(SOURCE_VOLUME_PATH_KEY)):
define_bronze_table(bronze_table)
dp.table is what registers a table with the pipeline. Here it’s called as a function rather than a decorator because the tables are generated in a loop. We’ll see the other syntax below. define_bronze_table takes the definition as an argument so that each closure binds its own entity. Together these two files declaratively define the Bronze layer.
Silver
Silver is the first layer where data is transformed from the Sleeper API schema. This layer has four steps: deduplication, joining split decimals, building relationships, and flattening arrays. In our configuration files, we define this logic using Spark code in dp.table declarations:
src/sleeper_lakehouse/silver_config.py
"""Table names, deduplication keys, and conforming rules for the Silver layer"""
from dataclasses import dataclass
FETCHED_AT_COLUMN = '_fetched_at'
SILVER_LEAGUES = 'leagues'
SILVER_TEAMS = 'teams'
SILVER_MATCHUPS = 'matchups'
SILVER_STARTING_LINEUPS = 'starting_lineups'
SILVER_PLAYOFF_RESULTS = 'playoff_results'
SILVER_TRANSACTIONS = 'transactions'
SILVER_PLAYERS = 'players'
SILVER_ROSTERED_PLAYERS = 'rostered_players'
SILVER_DRAFTS = 'drafts'
SILVER_DRAFT_PICKS = 'draft_picks'
SILVER_TABLES: tuple[str, ...] = (
SILVER_LEAGUES,
SILVER_TEAMS,
SILVER_MATCHUPS,
SILVER_STARTING_LINEUPS,
SILVER_PLAYOFF_RESULTS,
SILVER_TRANSACTIONS,
SILVER_PLAYERS,
SILVER_ROSTERED_PLAYERS,
SILVER_DRAFTS,
SILVER_DRAFT_PICKS,
)
SILVER_COMMENTS: dict[str, str] = {
SILVER_LEAGUES: 'One row per season of the league chain, with playoff configuration resolved',
SILVER_TEAMS: 'One row per team per season, joined to the managing user and with split points combined',
SILVER_MATCHUPS: 'One row per team per week with the opponent and result resolved',
SILVER_STARTING_LINEUPS: 'One row per started player per team per week',
SILVER_PLAYOFF_RESULTS: 'Final placement for every team that appears in a bracket',
SILVER_TRANSACTIONS: 'Typed waiver, free agent, and trade activity',
SILVER_PLAYERS: 'Current NFL player dimension limited to the columns worth carrying forward',
SILVER_ROSTERED_PLAYERS: 'One row per rostered player per team per week',
SILVER_DRAFTS: 'One row per draft, with the format, budget, and roster shape resolved',
SILVER_DRAFT_PICKS: 'One row per pick with the auction amount typed and keeper picks flagged',
}
DEDUPLICATION_KEYS: dict[str, tuple[str, ...]] = {
SILVER_LEAGUES: ('league_id',),
SILVER_TEAMS: ('league_id', 'roster_id'),
SILVER_MATCHUPS: ('league_id', 'week', 'roster_id'),
SILVER_STARTING_LINEUPS: ('league_id', 'week', 'roster_id', 'slot_index'),
SILVER_PLAYOFF_RESULTS: ('league_id', 'roster_id'),
SILVER_TRANSACTIONS: ('transaction_id',),
SILVER_PLAYERS: ('player_id',),
SILVER_ROSTERED_PLAYERS: ('league_id', 'week', 'roster_id', 'player_id'),
SILVER_DRAFTS: ('draft_id',),
SILVER_DRAFT_PICKS: ('draft_id', 'pick_number'),
}
EMPTY_ROSTER_SLOT = '0'
AUCTION_DRAFT_TYPE = 'auction'
@dataclass(frozen=True)
class SilverTable:
"""One conformed Silver table built from the Bronze layer"""
name: str
comment: str
deduplication_key: tuple[str, ...]
def qualified_name(catalog: str, schema: str, table: str) -> str:
"""Return a fully qualified table identifier"""
for part_name, part in (('catalog', catalog), ('schema', schema), ('table', table)):
if not isinstance(part, str) or not part.strip():
raise ValueError(f'{part_name} must be a non-empty string but received {part!r}')
if '.' in part:
raise ValueError(f'{part_name} must not contain a period but received {part!r}')
return f'{catalog}.{schema}.{table}'
def combine_split_points(whole_column: str, decimal_column: str) -> str:
"""Return the expression rebuilding a Sleeper score from its integer and decimal halves"""
return f'{whole_column} + coalesce({decimal_column}, 0) / 100.0'
def overall_place(bracket_place: int, is_consolation: bool, playoff_teams: int) -> int:
"""Return the league-wide finish for a placement decided inside one bracket"""
if isinstance(bracket_place, bool) or not isinstance(bracket_place, int):
raise ValueError(f'bracket_place must be an integer but received {bracket_place!r}')
if bracket_place < 1:
raise ValueError(f'bracket_place must be 1 or greater but received {bracket_place}')
if isinstance(playoff_teams, bool) or not isinstance(playoff_teams, int):
raise ValueError(f'playoff_teams must be an integer but received {playoff_teams!r}')
if playoff_teams < 1:
raise ValueError(f'playoff_teams must be 1 or greater but received {playoff_teams}')
return bracket_place + playoff_teams if is_consolation else bracket_place
def overall_place_expression(bracket_place_column: str, consolation_column: str, playoff_teams_column: str) -> str:
"""Return the expression form of the placement rule that overall_place pins down"""
return f'{bracket_place_column} + (CASE WHEN {consolation_column} THEN {playoff_teams_column} ELSE 0 END)'
def build_silver_tables() -> tuple[SilverTable, ...]:
"""Return the definition of every Silver table"""
return tuple(
SilverTable(
name=table,
comment=SILVER_COMMENTS[table],
deduplication_key=DEDUPLICATION_KEYS[table],
)
for table in SILVER_TABLES
)
src/sleeper_lakehouse_etl/transformations/silver_sleeper_entities.py
"""Silver tables conformed and deduplicated from the Bronze Sleeper entities"""
from utilities.project_path import add_project_package_to_path, pipeline_configuration
add_project_package_to_path()
from pyspark import pipelines as dp
from pyspark.sql import Column, DataFrame, Window
from pyspark.sql import functions as functions
from sleeper_lakehouse.silver_config import (
EMPTY_ROSTER_SLOT,
FETCHED_AT_COLUMN,
SILVER_COMMENTS,
SILVER_DRAFT_PICKS,
SILVER_DRAFTS,
SILVER_LEAGUES,
SILVER_MATCHUPS,
SILVER_PLAYERS,
SILVER_PLAYOFF_RESULTS,
SILVER_ROSTERED_PLAYERS,
SILVER_STARTING_LINEUPS,
SILVER_TEAMS,
SILVER_TRANSACTIONS,
combine_split_points,
overall_place_expression,
qualified_name,
)
CATALOG_KEY = 'sleeper.catalog'
SILVER_SCHEMA_KEY = 'sleeper.silver_schema'
CATALOG = pipeline_configuration(CATALOG_KEY)
SILVER_SCHEMA = pipeline_configuration(SILVER_SCHEMA_KEY)
def silver_name(table: str) -> str:
"""Return the fully qualified name of one Silver table"""
return qualified_name(CATALOG, SILVER_SCHEMA, table)
def latest_by(frame: DataFrame, key_columns: list[str]) -> DataFrame:
"""Keep only the most recently fetched row for each natural key, because Bronze is append-only"""
ordering = Window.partitionBy(*key_columns).orderBy(functions.col(FETCHED_AT_COLUMN).desc())
return (
frame.withColumn('_row_number', functions.row_number().over(ordering))
.filter(functions.col('_row_number') == 1)
.drop('_row_number')
)
def season_column() -> Column:
"""Return the landed season stamped by the extract job, typed as an integer"""
return functions.col('_season').cast('int').alias('season')
@dp.table(name=silver_name(SILVER_LEAGUES), comment=SILVER_COMMENTS[SILVER_LEAGUES])
def silver_leagues() -> DataFrame:
league = latest_by(spark.read.table('league'), ['_league_id'])
return league.select(
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('name').alias('league_name'),
functions.col('status'),
functions.col('sport'),
functions.col('total_rosters').cast('int').alias('total_rosters'),
functions.col('previous_league_id'),
functions.col('settings.playoff_week_start').cast('int').alias('playoff_week_start'),
functions.col('settings.playoff_teams').cast('int').alias('playoff_teams'),
functions.col('roster_positions'),
functions.col('scoring_settings'),
functions.col(FETCHED_AT_COLUMN),
)
@dp.table(name=silver_name(SILVER_TEAMS), comment=SILVER_COMMENTS[SILVER_TEAMS])
def silver_teams() -> DataFrame:
rosters = latest_by(spark.read.table('rosters'), ['_league_id', 'roster_id'])
users = latest_by(spark.read.table('users'), ['_league_id', 'user_id']).select(
functions.col('_league_id').alias('user_league_id'),
functions.col('user_id'),
functions.col('display_name').alias('manager_display_name'),
functions.col('is_bot').alias('manager_is_bot'),
)
joined = rosters.join(
users,
(rosters['_league_id'] == users['user_league_id']) & (rosters['owner_id'] == users['user_id']),
'left',
)
return joined.select(
rosters['_league_id'].alias('league_id'),
season_column(),
rosters['roster_id'].cast('int').alias('roster_id'),
rosters['owner_id'],
functions.col('manager_display_name'),
functions.coalesce(functions.col('manager_is_bot'), functions.lit(False)).alias('manager_is_bot'),
functions.coalesce(rosters['settings.wins'], functions.lit(0)).cast('int').alias('wins'),
functions.coalesce(rosters['settings.losses'], functions.lit(0)).cast('int').alias('losses'),
functions.coalesce(rosters['settings.ties'], functions.lit(0)).cast('int').alias('ties'),
functions.expr(combine_split_points('settings.fpts', 'settings.fpts_decimal')).alias('points_for'),
functions.expr(combine_split_points('settings.fpts_against', 'settings.fpts_against_decimal')).alias(
'points_against'
),
functions.expr(combine_split_points('settings.ppts', 'settings.ppts_decimal')).alias('maximum_points'),
rosters['settings.waiver_position'].cast('int').alias('waiver_position'),
rosters['settings.waiver_budget_used'].cast('int').alias('waiver_budget_used'),
rosters['settings.total_moves'].cast('int').alias('total_moves'),
rosters['settings.division'].cast('int').alias('division'),
rosters[FETCHED_AT_COLUMN],
)
@dp.table(name=silver_name(SILVER_MATCHUPS), comment=SILVER_COMMENTS[SILVER_MATCHUPS])
def silver_matchups() -> DataFrame:
matchups = latest_by(spark.read.table('matchups'), ['_league_id', '_week', 'roster_id']).select(
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('_week').cast('int').alias('week'),
functions.col('roster_id').cast('int').alias('roster_id'),
functions.col('matchup_id').cast('int').alias('matchup_id'),
functions.col('points'),
functions.col(FETCHED_AT_COLUMN),
)
opponents = matchups.select(
functions.col('league_id').alias('opponent_league_id'),
functions.col('week').alias('opponent_week'),
functions.col('matchup_id').alias('opponent_matchup_id'),
functions.col('roster_id').alias('opponent_roster_id'),
functions.col('points').alias('opponent_points'),
)
joined = matchups.join(
opponents,
(matchups['league_id'] == opponents['opponent_league_id'])
& (matchups['week'] == opponents['opponent_week'])
& (matchups['matchup_id'] == opponents['opponent_matchup_id'])
& (matchups['roster_id'] != opponents['opponent_roster_id']),
'left',
)
return joined.select(
matchups['league_id'],
matchups['season'],
matchups['week'],
matchups['roster_id'],
matchups['matchup_id'],
matchups['points'],
functions.col('opponent_roster_id').cast('int').alias('opponent_roster_id'),
functions.col('opponent_points'),
functions.when(functions.col('opponent_points').isNull(), functions.lit(None).cast('string'))
.when(matchups['points'] > functions.col('opponent_points'), functions.lit('win'))
.when(matchups['points'] < functions.col('opponent_points'), functions.lit('loss'))
.otherwise(functions.lit('tie'))
.alias('result'),
matchups[FETCHED_AT_COLUMN],
)
@dp.table(name=silver_name(SILVER_STARTING_LINEUPS), comment=SILVER_COMMENTS[SILVER_STARTING_LINEUPS])
def silver_starting_lineups() -> DataFrame:
matchups = latest_by(spark.read.table('matchups'), ['_league_id', '_week', 'roster_id'])
exploded = matchups.select(
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('_week').cast('int').alias('week'),
functions.col('roster_id').cast('int').alias('roster_id'),
functions.col('starters_points'),
functions.col(FETCHED_AT_COLUMN),
functions.posexplode(functions.col('starters')).alias('slot_index', 'player_id'),
)
return exploded.filter(functions.col('player_id') != functions.lit(EMPTY_ROSTER_SLOT)).select(
'league_id',
'season',
'week',
'roster_id',
'slot_index',
'player_id',
functions.col('starters_points').getItem(functions.col('slot_index')).alias('points'),
FETCHED_AT_COLUMN,
)
@dp.table(name=silver_name(SILVER_ROSTERED_PLAYERS), comment=SILVER_COMMENTS[SILVER_ROSTERED_PLAYERS])
def silver_rostered_players() -> DataFrame:
"""Every player on a roster in a week"""
matchups = latest_by(spark.read.table('matchups'), ['_league_id', '_week', 'roster_id'])
exploded = matchups.select(
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('_week').cast('int').alias('week'),
functions.col('roster_id').cast('int').alias('roster_id'),
functions.col('starters'),
functions.col(FETCHED_AT_COLUMN),
functions.explode(functions.col('players_points')).alias('player_id', 'points'),
)
return exploded.filter(functions.col('player_id') != functions.lit(EMPTY_ROSTER_SLOT)).select(
'league_id',
'season',
'week',
'roster_id',
'player_id',
'points',
functions.coalesce(
functions.array_contains(functions.col('starters'), functions.col('player_id')),
functions.lit(False),
).alias('was_started'),
FETCHED_AT_COLUMN,
)
@dp.table(name=silver_name(SILVER_PLAYOFF_RESULTS), comment=SILVER_COMMENTS[SILVER_PLAYOFF_RESULTS])
def silver_playoff_results() -> DataFrame:
def placements(table_name: str, is_consolation: bool) -> DataFrame:
bracket = spark.read.table(table_name).filter(functions.col('p').isNotNull())
winners = bracket.filter(functions.col('w').isNotNull()).select(
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('w').cast('int').alias('roster_id'),
functions.col('p').cast('int').alias('bracket_place'),
functions.lit(is_consolation).alias('is_consolation'),
functions.col(FETCHED_AT_COLUMN),
)
losers = bracket.filter(functions.col('l').isNotNull()).select(
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('l').cast('int').alias('roster_id'),
(functions.col('p').cast('int') + functions.lit(1)).alias('bracket_place'),
functions.lit(is_consolation).alias('is_consolation'),
functions.col(FETCHED_AT_COLUMN),
)
return winners.unionByName(losers)
bracket_places = placements('winners_bracket', False).unionByName(placements('losers_bracket', True))
leagues = spark.read.table(silver_name(SILVER_LEAGUES)).select(
functions.col('league_id').alias('league_league_id'),
functions.col('playoff_teams'),
)
joined = bracket_places.join(
leagues, bracket_places['league_id'] == leagues['league_league_id'], 'left'
).withColumn(
'final_place',
functions.expr(overall_place_expression('bracket_place', 'is_consolation', 'playoff_teams')).cast('int'),
)
return latest_by(joined, ['league_id', 'roster_id']).select(
'league_id',
'season',
'roster_id',
'bracket_place',
'is_consolation',
'final_place',
FETCHED_AT_COLUMN,
)
@dp.table(name=silver_name(SILVER_TRANSACTIONS), comment=SILVER_COMMENTS[SILVER_TRANSACTIONS])
def silver_transactions() -> DataFrame:
transactions = latest_by(spark.read.table('transactions'), ['transaction_id'])
return transactions.select(
functions.col('transaction_id'),
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('_week').cast('int').alias('week'),
functions.col('type').alias('transaction_type'),
functions.col('status'),
functions.to_timestamp(functions.col('created') / functions.lit(1000)).alias('created_at'),
functions.col('creator').alias('creator_user_id'),
functions.col('roster_ids'),
functions.col('adds'),
functions.col('drops'),
functions.col('settings.waiver_bid').cast('int').alias('waiver_bid'),
functions.col(FETCHED_AT_COLUMN),
)
@dp.table(name=silver_name(SILVER_PLAYERS), comment=SILVER_COMMENTS[SILVER_PLAYERS])
def silver_players() -> DataFrame:
players = latest_by(spark.read.table('players'), ['player_id'])
return players.select(
functions.col('player_id'),
functions.col('full_name'),
functions.col('first_name'),
functions.col('last_name'),
functions.col('position'),
functions.col('fantasy_positions'),
functions.col('team'),
functions.col('status'),
functions.col('active'),
functions.col('age').cast('int').alias('age'),
functions.col('years_exp').cast('int').alias('years_experience'),
functions.col('college'),
functions.col('birth_date'),
functions.col('injury_status'),
functions.col('number').cast('int').alias('jersey_number'),
functions.col('search_rank').cast('long').alias('search_rank'),
functions.col(FETCHED_AT_COLUMN),
)
@dp.table(name=silver_name(SILVER_DRAFTS), comment=SILVER_COMMENTS[SILVER_DRAFTS])
def silver_drafts() -> DataFrame:
drafts = latest_by(spark.read.table('drafts'), ['draft_id'])
return drafts.select(
functions.col('draft_id'),
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('type').alias('draft_type'),
functions.col('status'),
functions.col('settings.budget').cast('int').alias('budget'),
functions.col('settings.teams').cast('int').alias('teams'),
functions.col('settings.rounds').cast('int').alias('rounds'),
functions.to_timestamp(functions.col('start_time') / functions.lit(1000)).alias('started_at'),
functions.col('draft_order'),
functions.col(FETCHED_AT_COLUMN),
)
@dp.table(name=silver_name(SILVER_DRAFT_PICKS), comment=SILVER_COMMENTS[SILVER_DRAFT_PICKS])
def silver_draft_picks() -> DataFrame:
"""Silver draft picks"""
# the key is read against Bronze, which mirrors the API, so it is the source spelling rather than the
# conformed one this table goes on to publish
picks = latest_by(spark.read.table('draft_picks'), ['draft_id', 'pick_no'])
return picks.select(
functions.col('draft_id'),
functions.col('_league_id').alias('league_id'),
season_column(),
functions.col('pick_no').cast('int').alias('pick_number'),
functions.col('round').cast('int').alias('round'),
functions.col('draft_slot').cast('int').alias('draft_slot'),
functions.col('roster_id').cast('int').alias('roster_id'),
functions.expr("nullif(picked_by, '')").alias('picked_by_user_id'),
functions.col('player_id'),
functions.col('metadata').getItem('amount').cast('int').alias('amount'),
functions.coalesce(functions.col('is_keeper'), functions.lit(False)).alias('is_keeper'),
functions.col(FETCHED_AT_COLUMN),
)
Gold
Finally, Gold uses the silver layer to create our final tables for analytics:
src/sleeper_lakehouse/gold_config.py
"""Table names and derived metric expressions for the Gold layer"""
from dataclasses import dataclass
GOLD_SEASON_STANDINGS = 'season_standings'
GOLD_MANAGER_RECORDS = 'manager_records'
GOLD_HEAD_TO_HEAD = 'head_to_head'
GOLD_WEEKLY_SCORES = 'weekly_scores'
GOLD_DRAFT_SPEND = 'draft_spend'
GOLD_DRAFT_TEAM_SUMMARY = 'draft_team_summary'
GOLD_TABLES: tuple[str, ...] = (
GOLD_SEASON_STANDINGS,
GOLD_MANAGER_RECORDS,
GOLD_HEAD_TO_HEAD,
GOLD_WEEKLY_SCORES,
GOLD_DRAFT_SPEND,
GOLD_DRAFT_TEAM_SUMMARY,
)
GOLD_COMMENTS: dict[str, str] = {
GOLD_SEASON_STANDINGS: 'Final regular season record and playoff finish for every team in every season',
GOLD_MANAGER_RECORDS: 'Career totals for every manager across the whole league chain',
GOLD_HEAD_TO_HEAD: 'All-time record between every pair of managers that has met',
GOLD_WEEKLY_SCORES: 'Weekly scoring with league-wide rank and schedule-independent expected wins',
GOLD_DRAFT_SPEND: 'One row per auction pick',
GOLD_DRAFT_TEAM_SUMMARY: 'One row per team per auction',
}
CHAMPION_PLACE = 1
MINIMUM_AUCTION_BID = 1
POSITION_COLUMN_NAMES: dict[str, str] = {
'QB': 'quarterback',
'RB': 'running_back',
'WR': 'wide_receiver',
'TE': 'tight_end',
'K': 'kicker',
'DEF': 'defense',
}
@dataclass(frozen=True)
class GoldTable:
"""One curated Gold table built from the Silver layer"""
name: str
comment: str
def win_percentage_expression(wins_column: str, losses_column: str, ties_column: str) -> str:
"""Return the expression scoring a tie as half a win"""
games = f'{wins_column} + {losses_column} + {ties_column}'
return f'({wins_column} + 0.5 * {ties_column}) / nullif({games}, 0)'
def expected_wins_expression(weekly_rank_column: str, team_count_column: str) -> str:
"""Return the share of the league a team outscored in one week, ignoring who it was scheduled against"""
return f'({team_count_column} - {weekly_rank_column}) / nullif({team_count_column} - 1, 0)'
def made_playoffs_expression(final_place_column: str, playoff_teams_column: str) -> str:
"""Return whether a finish placed inside the championship bracket"""
return f'{final_place_column} IS NOT NULL AND {final_place_column} <= {playoff_teams_column}'
def budget_share_expression(amount_column: str, budget_column: str) -> str:
"""Return the fraction of a whole auction budget that one price consumed"""
return f'{amount_column} / nullif({budget_column}, 0)'
def points_per_dollar_expression(points_column: str, amount_column: str) -> str:
"""Return the points a dollar of auction spend returned, left null before any game has been played"""
return f'{points_column} / nullif({amount_column}, 0)'
def market_premium_expression(amount_column: str, position_median_column: str) -> str:
"""Return how far a price sat above what the same room paid for the typical player at that position"""
return f'{amount_column} - {position_median_column}'
def consensus_gap_expression(spend_rank_column: str, consensus_rank_column: str) -> str:
"""Return how many places above his consensus standing a player was paid"""
return f'{consensus_rank_column} - {spend_rank_column}'
def gini_expression(rank_weighted_spend_column: str, pick_count_column: str, total_spend_column: str) -> str:
"""Return the Gini coefficient of one team's prices"""
return (
f'2 * {rank_weighted_spend_column} / nullif({pick_count_column} * {total_spend_column}, 0)'
f' - ({pick_count_column} + 1) / nullif({pick_count_column}, 0)'
)
def build_gold_tables() -> tuple[GoldTable, ...]:
"""Return the definition of every Gold table"""
return tuple(GoldTable(name=table, comment=GOLD_COMMENTS[table]) for table in GOLD_TABLES)
src/sleeper_lakehouse_etl/transformations/gold_sleeper_analytics.py
"""Curated Gold tables answering league-wide questions from the Silver layer"""
from utilities.project_path import add_project_package_to_path, pipeline_configuration
add_project_package_to_path()
from pyspark import pipelines as dp
from pyspark.sql import Column, DataFrame, Window
from pyspark.sql import functions as functions
from sleeper_lakehouse.gold_config import (
CHAMPION_PLACE,
GOLD_COMMENTS,
GOLD_DRAFT_SPEND,
GOLD_DRAFT_TEAM_SUMMARY,
GOLD_HEAD_TO_HEAD,
GOLD_MANAGER_RECORDS,
GOLD_SEASON_STANDINGS,
GOLD_WEEKLY_SCORES,
MINIMUM_AUCTION_BID,
POSITION_COLUMN_NAMES,
budget_share_expression,
consensus_gap_expression,
expected_wins_expression,
gini_expression,
made_playoffs_expression,
market_premium_expression,
points_per_dollar_expression,
win_percentage_expression,
)
from sleeper_lakehouse.silver_config import (
AUCTION_DRAFT_TYPE,
SILVER_DRAFT_PICKS,
SILVER_DRAFTS,
SILVER_LEAGUES,
SILVER_MATCHUPS,
SILVER_PLAYERS,
SILVER_PLAYOFF_RESULTS,
SILVER_ROSTERED_PLAYERS,
SILVER_TEAMS,
qualified_name,
)
TOP_SPEND_PLAYER_COUNT = 3
CATALOG_KEY = 'sleeper.catalog'
SILVER_SCHEMA_KEY = 'sleeper.silver_schema'
GOLD_SCHEMA_KEY = 'sleeper.gold_schema'
CATALOG = pipeline_configuration(CATALOG_KEY)
SILVER_SCHEMA = pipeline_configuration(SILVER_SCHEMA_KEY)
GOLD_SCHEMA = pipeline_configuration(GOLD_SCHEMA_KEY)
def silver_name(table: str) -> str:
"""Return the fully qualified name of one Silver table"""
return qualified_name(CATALOG, SILVER_SCHEMA, table)
def gold_name(table: str) -> str:
"""Return the fully qualified name of one Gold table"""
return qualified_name(CATALOG, GOLD_SCHEMA, table)
@dp.table(name=gold_name(GOLD_SEASON_STANDINGS), comment=GOLD_COMMENTS[GOLD_SEASON_STANDINGS])
def gold_season_standings() -> DataFrame:
teams = spark.read.table(silver_name(SILVER_TEAMS))
leagues = spark.read.table(silver_name(SILVER_LEAGUES)).select(
functions.col('league_id').alias('league_league_id'),
functions.col('status').alias('season_status'),
functions.col('playoff_teams'),
functions.col('playoff_week_start'),
)
playoff_results = spark.read.table(silver_name(SILVER_PLAYOFF_RESULTS)).select(
functions.col('league_id').alias('playoff_league_id'),
functions.col('roster_id').alias('playoff_roster_id'),
functions.col('final_place'),
)
joined = teams.join(leagues, teams['league_id'] == leagues['league_league_id'], 'left').join(
playoff_results,
(teams['league_id'] == playoff_results['playoff_league_id'])
& (teams['roster_id'] == playoff_results['playoff_roster_id']),
'left',
)
regular_season_order = Window.partitionBy('league_id').orderBy(
functions.col('wins').desc(),
functions.col('points_for').desc(),
)
return (
joined.withColumn('regular_season_rank', functions.rank().over(regular_season_order))
.withColumn('win_percentage', functions.expr(win_percentage_expression('wins', 'losses', 'ties')))
.withColumn('made_playoffs', functions.expr(made_playoffs_expression('final_place', 'playoff_teams')))
.withColumn('is_champion', functions.col('final_place') == functions.lit(CHAMPION_PLACE))
.select(
'season',
'league_id',
'roster_id',
'owner_id',
'manager_display_name',
'season_status',
'wins',
'losses',
'ties',
'win_percentage',
'points_for',
'points_against',
'maximum_points',
'regular_season_rank',
'final_place',
'made_playoffs',
'is_champion',
'playoff_teams',
'playoff_week_start',
)
)
@dp.table(name=gold_name(GOLD_MANAGER_RECORDS), comment=GOLD_COMMENTS[GOLD_MANAGER_RECORDS])
def gold_manager_records() -> DataFrame:
standings = spark.read.table(gold_name(GOLD_SEASON_STANDINGS)).filter(functions.col('owner_id').isNotNull())
latest_name_order = Window.partitionBy('owner_id').orderBy(functions.col('season').desc())
with_latest_name = standings.withColumn(
'latest_display_name', functions.first('manager_display_name').over(latest_name_order)
)
return (
with_latest_name.groupBy('owner_id')
.agg(
functions.max('latest_display_name').alias('manager_display_name'),
functions.sum(((functions.col('wins') + functions.col('losses') + functions.col('ties')) > 0).cast('int'))
.cast('int')
.alias('seasons_played'),
functions.count(functions.lit(1)).cast('int').alias('seasons_in_league'),
functions.sum('wins').cast('int').alias('wins'),
functions.sum('losses').cast('int').alias('losses'),
functions.sum('ties').cast('int').alias('ties'),
functions.round(functions.sum('points_for'), 2).alias('career_points_for'),
functions.round(functions.sum('points_against'), 2).alias('career_points_against'),
functions.sum(functions.col('is_champion').cast('int')).cast('int').alias('championships'),
functions.sum(functions.col('made_playoffs').cast('int')).cast('int').alias('playoff_appearances'),
functions.round(functions.avg('final_place'), 2).alias('average_final_place'),
functions.min('final_place').cast('int').alias('best_finish'),
functions.max('final_place').cast('int').alias('worst_finish'),
)
.withColumn('win_percentage', functions.expr(win_percentage_expression('wins', 'losses', 'ties')))
.orderBy(functions.col('win_percentage').desc())
)
@dp.table(name=gold_name(GOLD_HEAD_TO_HEAD), comment=GOLD_COMMENTS[GOLD_HEAD_TO_HEAD])
def gold_head_to_head() -> DataFrame:
matchups = spark.read.table(silver_name(SILVER_MATCHUPS)).filter(functions.col('opponent_roster_id').isNotNull())
teams = spark.read.table(silver_name(SILVER_TEAMS)).select(
functions.col('league_id').alias('team_league_id'),
functions.col('roster_id').alias('team_roster_id'),
functions.col('owner_id'),
functions.col('manager_display_name'),
)
opponents = teams.select(
functions.col('team_league_id').alias('opponent_team_league_id'),
functions.col('team_roster_id').alias('opponent_team_roster_id'),
functions.col('owner_id').alias('opponent_owner_id'),
functions.col('manager_display_name').alias('opponent_display_name'),
)
joined = (
matchups.join(
teams,
(matchups['league_id'] == teams['team_league_id']) & (matchups['roster_id'] == teams['team_roster_id']),
'inner',
)
.join(
opponents,
(matchups['league_id'] == opponents['opponent_team_league_id'])
& (matchups['opponent_roster_id'] == opponents['opponent_team_roster_id']),
'inner',
)
.filter(functions.col('owner_id').isNotNull() & functions.col('opponent_owner_id').isNotNull())
)
return (
joined.groupBy('owner_id', 'opponent_owner_id')
.agg(
functions.max('manager_display_name').alias('manager_display_name'),
functions.max('opponent_display_name').alias('opponent_display_name'),
functions.count(functions.lit(1)).cast('int').alias('meetings'),
functions.sum((functions.col('result') == functions.lit('win')).cast('int')).cast('int').alias('wins'),
functions.sum((functions.col('result') == functions.lit('loss')).cast('int')).cast('int').alias('losses'),
functions.sum((functions.col('result') == functions.lit('tie')).cast('int')).cast('int').alias('ties'),
functions.round(functions.sum('points'), 2).alias('points_for'),
functions.round(functions.sum('opponent_points'), 2).alias('points_against'),
)
.withColumn('win_percentage', functions.expr(win_percentage_expression('wins', 'losses', 'ties')))
)
@dp.table(name=gold_name(GOLD_WEEKLY_SCORES), comment=GOLD_COMMENTS[GOLD_WEEKLY_SCORES])
def gold_weekly_scores() -> DataFrame:
matchups = spark.read.table(silver_name(SILVER_MATCHUPS))
teams = spark.read.table(silver_name(SILVER_TEAMS)).select(
functions.col('league_id').alias('team_league_id'),
functions.col('roster_id').alias('team_roster_id'),
functions.col('owner_id'),
functions.col('manager_display_name'),
)
leagues = spark.read.table(silver_name(SILVER_LEAGUES)).select(
functions.col('league_id').alias('league_league_id'),
functions.col('playoff_week_start'),
)
joined = matchups.join(
teams,
(matchups['league_id'] == teams['team_league_id']) & (matchups['roster_id'] == teams['team_roster_id']),
'left',
).join(leagues, matchups['league_id'] == leagues['league_league_id'], 'left')
weekly_order = Window.partitionBy('league_id', 'week').orderBy(functions.col('points').desc())
weekly_partition = Window.partitionBy('league_id', 'week')
return (
joined.withColumn('weekly_rank', functions.rank().over(weekly_order))
.withColumn('team_count', functions.count(functions.lit(1)).over(weekly_partition))
.withColumn('expected_wins', functions.expr(expected_wins_expression('weekly_rank', 'team_count')))
.withColumn('is_regular_season', functions.col('week') < functions.col('playoff_week_start'))
.select(
'season',
'league_id',
'week',
'roster_id',
'owner_id',
'manager_display_name',
'points',
'opponent_roster_id',
'opponent_points',
'result',
functions.col('weekly_rank').cast('int').alias('weekly_rank'),
functions.col('team_count').cast('int').alias('team_count'),
'expected_wins',
'is_regular_season',
)
)
def auction_picks() -> DataFrame:
"""Every pick from a draft Sleeper recorded as an auction"""
picks = spark.read.table(silver_name(SILVER_DRAFT_PICKS))
drafts = (
spark.read.table(silver_name(SILVER_DRAFTS))
.filter(functions.col('draft_type') == functions.lit(AUCTION_DRAFT_TYPE))
.select(
functions.col('draft_id').alias('auction_draft_id'),
functions.col('draft_type'),
functions.col('budget'),
)
)
teams = spark.read.table(silver_name(SILVER_TEAMS)).select(
functions.col('league_id').alias('team_league_id'),
functions.col('roster_id').alias('team_roster_id'),
functions.col('owner_id'),
functions.col('manager_display_name'),
)
players = spark.read.table(silver_name(SILVER_PLAYERS)).select(
functions.col('player_id').alias('dimension_player_id'),
# Sleeper leaves full_name empty for a team defense and splits the city and nickname across the two
# name columns instead, so a drafted defense would otherwise come through unnamed
functions.coalesce(
functions.col('full_name'),
functions.concat_ws(' ', functions.col('first_name'), functions.col('last_name')),
).alias('player_name'),
functions.col('position'),
functions.col('team').alias('nfl_team'),
functions.col('search_rank'),
)
return (
picks.join(drafts, picks['draft_id'] == drafts['auction_draft_id'], 'inner')
.join(
teams,
(picks['league_id'] == teams['team_league_id']) & (picks['roster_id'] == teams['team_roster_id']),
'left',
)
.join(players, picks['player_id'] == players['dimension_player_id'], 'left')
.drop('auction_draft_id', 'team_league_id', 'team_roster_id', 'dimension_player_id')
)
def drafted_player_points() -> DataFrame:
"""Total the points a player scored while sitting on one roster"""
rostered = spark.read.table(silver_name(SILVER_ROSTERED_PLAYERS))
return (
rostered.groupBy('league_id', 'roster_id', 'player_id')
.agg(
functions.round(functions.sum('points'), 2).alias('points_for_drafting_team'),
functions.count(functions.lit(1)).cast('int').alias('weeks_rostered'),
functions.sum(functions.col('was_started').cast('int')).cast('int').alias('weeks_started'),
)
.select(
functions.col('league_id').alias('points_league_id'),
functions.col('roster_id').alias('points_roster_id'),
functions.col('player_id').alias('points_player_id'),
'points_for_drafting_team',
'weeks_rostered',
'weeks_started',
)
)
def _amount_above_the_minimum_bid() -> Column:
"""Return the amount only for a pick that cleared the minimum bid, so a baseline that ignores the dollar
tail is one aggregate over the same frame rather than a second grouping joined back in"""
return functions.when(functions.col('amount') > functions.lit(MINIMUM_AUCTION_BID), functions.col('amount'))
@dp.table(name=gold_name(GOLD_DRAFT_SPEND), comment=GOLD_COMMENTS[GOLD_DRAFT_SPEND])
def gold_draft_spend() -> DataFrame:
"""Judging spend against what the same room paid for the typical player at that position, and against
Sleeper's own ordering of players, with each positional baseline offered four ways because the median over
every pick is dragged toward the floor by the dollar fliers that fill out a roster"""
picks = auction_picks()
position_baseline = (
picks.filter(~functions.col('is_keeper'))
.groupBy('league_id', 'position')
.agg(
functions.percentile_approx('amount', 0.5).cast('double').alias('position_median_amount'),
functions.percentile_approx(_amount_above_the_minimum_bid(), 0.5)
.cast('double')
.alias('position_median_amount_above_minimum'),
functions.avg('amount').cast('double').alias('position_mean_amount'),
functions.avg(_amount_above_the_minimum_bid()).cast('double').alias('position_mean_amount_above_minimum'),
)
.select(
functions.col('league_id').alias('baseline_league_id'),
functions.col('position').alias('baseline_position'),
'position_median_amount',
'position_median_amount_above_minimum',
'position_mean_amount',
'position_mean_amount_above_minimum',
)
)
points = drafted_player_points()
joined = picks.join(
position_baseline,
(picks['league_id'] == position_baseline['baseline_league_id'])
& (picks['position'].eqNullSafe(position_baseline['baseline_position'])),
'left',
).join(
points,
(picks['league_id'] == points['points_league_id'])
& (picks['roster_id'] == points['points_roster_id'])
& (picks['player_id'] == points['points_player_id']),
'left',
)
position_spend_order = Window.partitionBy('league_id', 'position').orderBy(functions.col('amount').desc())
overall_spend_order = Window.partitionBy('league_id').orderBy(functions.col('amount').desc())
consensus_order = Window.partitionBy('league_id', 'position').orderBy(functions.col('search_rank').asc_nulls_last())
return (
joined.withColumn('position_spend_rank', functions.rank().over(position_spend_order))
.withColumn('overall_spend_rank', functions.rank().over(overall_spend_order))
.withColumn('consensus_position_rank', functions.rank().over(consensus_order))
.withColumn('percent_of_budget', functions.expr(budget_share_expression('amount', 'budget')))
.withColumn('market_premium', functions.expr(market_premium_expression('amount', 'position_median_amount')))
.withColumn(
'market_premium_above_minimum',
functions.expr(market_premium_expression('amount', 'position_median_amount_above_minimum')),
)
.withColumn(
'market_premium_versus_mean',
functions.expr(market_premium_expression('amount', 'position_mean_amount')),
)
.withColumn(
'market_premium_versus_mean_above_minimum',
functions.expr(market_premium_expression('amount', 'position_mean_amount_above_minimum')),
)
.withColumn(
'consensus_gap',
functions.expr(consensus_gap_expression('position_spend_rank', 'consensus_position_rank')),
)
.withColumn(
'points_per_dollar',
functions.expr(points_per_dollar_expression('points_for_drafting_team', 'amount')),
)
.select(
'season',
'league_id',
'draft_id',
'roster_id',
'owner_id',
'manager_display_name',
'player_id',
'player_name',
'position',
'nfl_team',
'pick_number',
'round',
'amount',
'budget',
'percent_of_budget',
'is_keeper',
'position_median_amount',
'position_median_amount_above_minimum',
'position_mean_amount',
'position_mean_amount_above_minimum',
'market_premium',
'market_premium_above_minimum',
'market_premium_versus_mean',
'market_premium_versus_mean_above_minimum',
functions.col('position_spend_rank').cast('int').alias('position_spend_rank'),
functions.col('overall_spend_rank').cast('int').alias('overall_spend_rank'),
functions.col('consensus_position_rank').cast('int').alias('consensus_position_rank'),
'consensus_gap',
'search_rank',
'points_for_drafting_team',
'weeks_rostered',
'weeks_started',
'points_per_dollar',
)
)
@dp.table(name=gold_name(GOLD_DRAFT_TEAM_SUMMARY), comment=GOLD_COMMENTS[GOLD_DRAFT_TEAM_SUMMARY])
def gold_draft_team_summary() -> DataFrame:
"""Summarize draft"""
spend = spark.read.table(gold_name(GOLD_DRAFT_SPEND))
cheapest_first = Window.partitionBy('league_id', 'roster_id').orderBy(functions.col('amount').asc())
priciest_first = Window.partitionBy('league_id', 'roster_id').orderBy(functions.col('amount').desc())
ranked = spend.withColumn('cheapest_first_rank', functions.row_number().over(cheapest_first)).withColumn(
'priciest_first_rank', functions.row_number().over(priciest_first)
)
def spend_on(position: str) -> Column:
"""A team that drafted nobody at a position spent zero there, not an unknown amount, so the sum falls
back rather than returning the null that an empty set would otherwise produce"""
return (
functions.sum(
functions.when(functions.col('position') == functions.lit(position), functions.col('amount')).otherwise(
0
)
)
.cast('int')
.alias(f'spend_{POSITION_COLUMN_NAMES[position]}')
)
return (
ranked.groupBy('season', 'league_id', 'draft_id', 'roster_id', 'owner_id', 'manager_display_name', 'budget')
.agg(
functions.sum('amount').cast('int').alias('total_spent'),
functions.count(functions.lit(1)).cast('int').alias('players_drafted'),
functions.sum(functions.col('is_keeper').cast('int')).cast('int').alias('keepers_drafted'),
functions.sum(functions.when(functions.col('is_keeper'), functions.col('amount')).otherwise(0))
.cast('int')
.alias('keeper_spend'),
functions.sum(functions.when(~functions.col('is_keeper'), functions.col('amount')).otherwise(0))
.cast('int')
.alias('open_market_spend'),
functions.max('amount').cast('int').alias('maximum_price'),
functions.percentile_approx('amount', 0.5).cast('double').alias('median_price'),
functions.sum(
functions.when(functions.col('priciest_first_rank') == functions.lit(1), functions.col('amount'))
)
.cast('int')
.alias('top_player_spend'),
functions.sum(
functions.when(
functions.col('priciest_first_rank') <= functions.lit(TOP_SPEND_PLAYER_COUNT),
functions.col('amount'),
)
)
.cast('int')
.alias('top_player_group_spend'),
functions.sum(functions.col('cheapest_first_rank') * functions.col('amount')).alias('rank_weighted_spend'),
spend_on('QB'),
spend_on('RB'),
spend_on('WR'),
spend_on('TE'),
spend_on('K'),
spend_on('DEF'),
functions.round(functions.sum('points_for_drafting_team'), 2).alias('points_from_draft'),
)
.withColumn('budget_remaining', functions.col('budget') - functions.col('total_spent'))
.withColumn('top_player_share', functions.expr('top_player_spend / nullif(total_spent, 0)'))
.withColumn('top_player_group_share', functions.expr('top_player_group_spend / nullif(total_spent, 0)'))
.withColumn(
'spend_gini',
functions.expr(gini_expression('rank_weighted_spend', 'players_drafted', 'total_spent')),
)
.withColumn(
'points_per_dollar',
functions.expr(points_per_dollar_expression('points_from_draft', 'total_spent')),
)
.drop('rank_weighted_spend', 'top_player_spend', 'top_player_group_spend')
)
Scheduling the Daily Extract
The backfill runs once. After that, this job keeps the lakehouse up to date by landing only the current season, since completed seasons never change.
resources/sleeper_extract.job.yml
resources:
jobs:
sleeper_extract:
name: sleeper_extract
description: >-
Lands the current season from the Sleeper API everyday
trigger:
periodic:
interval: 1
unit: DAYS
parameters:
- name: seasons
default: current
- name: force
default: "false"
tasks:
- task_key: extract
python_wheel_task:
package_name: sleeper_lakehouse
entry_point: extract
parameters:
- "--volume-path"
- /Volumes/${var.catalog}/${resources.schemas.raw.name}/${resources.volumes.source.name}
- "--root-league-id"
- "${var.root_league_id}"
- "--sport"
- "${var.sport}"
- "--seasons"
- "{{job.parameters.seasons}}"
- "--force"
- "{{job.parameters.force}}"
environment_key: default
- task_key: refresh_lakehouse
depends_on:
- task_key: extract
pipeline_task:
pipeline_id: ${resources.pipelines.sleeper_lakehouse_etl.id}
environments:
- environment_key: default
spec:
environment_version: "4"
dependencies:
- ../dist/*.whl
The Notebook and Dashboard
Two resources round out the bundle. The notebook is deployed as a job:
resources/sleeper_lakehouse_tour.job.yml
resources:
jobs:
lakehouse_tour:
name: lakehouse_tour
description: >-
Guided walkthrough of every layer of the lakehouse. Open the notebook to read it, or run
the job to execute every cell against the deployed tables.
parameters:
- name: catalog
default: ${var.catalog}
- name: raw_schema
default: ${resources.schemas.raw.name}
- name: bronze_schema
default: ${resources.schemas.bronze.name}
- name: silver_schema
default: ${resources.schemas.silver.name}
- name: gold_schema
default: ${resources.schemas.gold.name}
- name: source_volume
default: ${var.source_volume}
tasks:
- task_key: tour
notebook_task:
notebook_path: ../src/notebooks/lakehouse_tour.ipynb
The dashboard has a dedicated resource type:
resources/sleeper_dashboard.yml
resources:
dashboards:
league_overview:
display_name: Sleeper League Overview
file_path: ../src/dashboards/league_overview.lvdash.json
warehouse_id: ${var.warehouse_id}
That definition file holds both the SQL statements and the widget layout:
src/dashboards/league_overview.lvdash.json
{
"datasets": [
{
"name": "manager_records",
"displayName": "All-time manager records",
"queryLines": [
"SELECT * FROM ${resources.schemas.gold.catalog_name}.${resources.schemas.gold.name}.manager_records ORDER BY win_percentage DESC"
]
},
{
"name": "season_standings",
"displayName": "Season standings",
"queryLines": [
"SELECT * FROM ${resources.schemas.gold.catalog_name}.${resources.schemas.gold.name}.season_standings"
]
}
],
"pages": [
{
"name": "league_overview",
"displayName": "League Overview",
"layout": [
{
"widget": {
"name": "managers_counter",
"queries": [
{
"name": "main_query",
"query": {
"datasetName": "manager_records",
"fields": [
{
"name": "count(owner_id)",
"expression": "COUNT(`owner_id`)"
}
],
"disaggregated": false
}
}
],
"spec": {
"version": 2,
"widgetType": "counter",
"frame": {
"showDescription": false,
"showTitle": true,
"title": "Managers"
},
"encodings": {
"value": {
"fieldName": "count(owner_id)",
"displayName": "Managers"
}
}
}
},
"position": {
"x": 0,
"y": 0,
"width": 2,
"height": 3
}
},
{
"widget": {
"name": "seasons_counter",
"queries": [
{
"name": "main_query",
"query": {
"datasetName": "season_standings",
"fields": [
{
"name": "count(season)",
"expression": "COUNT(DISTINCT `season`)"
}
],
"disaggregated": false
}
}
],
"spec": {
"version": 2,
"widgetType": "counter",
"frame": {
"showDescription": false,
"showTitle": true,
"title": "Seasons"
},
"encodings": {
"value": {
"fieldName": "count(season)",
"displayName": "Seasons"
}
}
}
},
"position": {
"x": 2,
"y": 0,
"width": 2,
"height": 3
}
},
{
"widget": {
"name": "titles_counter",
"queries": [
{
"name": "main_query",
"query": {
"datasetName": "manager_records",
"fields": [
{
"name": "sum(championships)",
"expression": "SUM(`championships`)"
}
],
"disaggregated": false
}
}
],
"spec": {
"version": 2,
"widgetType": "counter",
"frame": {
"showDescription": false,
"showTitle": true,
"title": "Championships"
},
"encodings": {
"value": {
"fieldName": "sum(championships)",
"displayName": "Championships"
}
}
}
},
"position": {
"x": 4,
"y": 0,
"width": 2,
"height": 3
}
},
{
"widget": {
"name": "manager_records_table",
"queries": [
{
"name": "main_query",
"query": {
"datasetName": "manager_records",
"fields": [
{
"name": "manager_display_name",
"expression": "`manager_display_name`"
},
{
"name": "seasons_played",
"expression": "`seasons_played`"
},
{
"name": "wins",
"expression": "`wins`"
},
{
"name": "losses",
"expression": "`losses`"
},
{
"name": "win_percentage",
"expression": "`win_percentage`"
},
{
"name": "championships",
"expression": "`championships`"
},
{
"name": "playoff_appearances",
"expression": "`playoff_appearances`"
},
{
"name": "average_final_place",
"expression": "`average_final_place`"
},
{
"name": "career_points_for",
"expression": "`career_points_for`"
}
],
"disaggregated": false
}
}
],
"spec": {
"allowHTMLByDefault": false,
"condensed": false,
"frame": {
"showDescription": false,
"showTitle": true,
"title": "All-time manager records"
},
"invisibleColumns": [],
"itemsPerPage": 12,
"paginationSize": "default",
"version": 1,
"widgetType": "table",
"withRowNumber": false,
"encodings": {
"columns": [
{
"alignContent": "left",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "string",
"displayName": "Manager",
"fieldName": "manager_display_name",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10000,
"preserveWhitespace": false,
"title": "Manager",
"type": "string",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "Seasons",
"fieldName": "seasons_played",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10001,
"preserveWhitespace": false,
"title": "Seasons",
"type": "integer",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "W",
"fieldName": "wins",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10002,
"preserveWhitespace": false,
"title": "W",
"type": "integer",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "L",
"fieldName": "losses",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10003,
"preserveWhitespace": false,
"title": "L",
"type": "integer",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "Win %",
"fieldName": "win_percentage",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10004,
"preserveWhitespace": false,
"title": "Win %",
"type": "float",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "Titles",
"fieldName": "championships",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10005,
"preserveWhitespace": false,
"title": "Titles",
"type": "integer",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "Playoffs",
"fieldName": "playoff_appearances",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10006,
"preserveWhitespace": false,
"title": "Playoffs",
"type": "integer",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "Avg Finish",
"fieldName": "average_final_place",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10007,
"preserveWhitespace": false,
"title": "Avg Finish",
"type": "float",
"useMonospaceFont": false,
"visible": true
},
{
"alignContent": "right",
"allowHTML": false,
"allowSearch": false,
"booleanValues": [
"false",
"true"
],
"dateTimeFormat": "YYYY-MM-DD",
"displayAs": "number",
"displayName": "Career PF",
"fieldName": "career_points_for",
"highlightLinks": false,
"imageHeight": "",
"imageTitleTemplate": "{{ @ }}",
"imageUrlTemplate": "{{ @ }}",
"imageWidth": "",
"linkOpenInNewTab": true,
"linkTextTemplate": "{{ @ }}",
"linkTitleTemplate": "{{ @ }}",
"linkUrlTemplate": "{{ @ }}",
"order": 10008,
"preserveWhitespace": false,
"title": "Career PF",
"type": "float",
"useMonospaceFont": false,
"visible": true
}
]
}
}
},
"position": {
"x": 0,
"y": 3,
"width": 6,
"height": 8
}
},
{
"widget": {
"name": "win_percentage_bar",
"queries": [
{
"name": "main_query",
"query": {
"datasetName": "manager_records",
"fields": [
{
"name": "manager_display_name",
"expression": "`manager_display_name`"
},
{
"name": "avg(win_percentage)",
"expression": "AVG(`win_percentage`)"
}
],
"disaggregated": false
}
}
],
"spec": {
"version": 3,
"widgetType": "bar",
"frame": {
"showDescription": false,
"showTitle": true,
"title": "Career win percentage"
},
"encodings": {
"x": {
"fieldName": "manager_display_name",
"scale": {
"type": "categorical"
},
"displayName": "Manager",
"axis": {
"hideTitle": true
}
},
"y": {
"fieldName": "avg(win_percentage)",
"scale": {
"type": "quantitative"
},
"displayName": "Win %",
"axis": {
"title": "Win %"
}
}
}
}
},
"position": {
"x": 0,
"y": 11,
"width": 3,
"height": 7
}
},
{
"widget": {
"name": "championships_bar",
"queries": [
{
"name": "main_query",
"query": {
"datasetName": "manager_records",
"fields": [
{
"name": "manager_display_name",
"expression": "`manager_display_name`"
},
{
"name": "sum(championships)",
"expression": "SUM(`championships`)"
}
],
"disaggregated": false
}
}
],
"spec": {
"version": 3,
"widgetType": "bar",
"frame": {
"showDescription": false,
"showTitle": true,
"title": "Championships won"
},
"encodings": {
"x": {
"fieldName": "manager_display_name",
"scale": {
"type": "categorical"
},
"displayName": "Manager",
"axis": {
"hideTitle": true
}
},
"y": {
"fieldName": "sum(championships)",
"scale": {
"type": "quantitative"
},
"displayName": "Championships",
"axis": {
"title": "Championships"
}
}
}
}
},
"position": {
"x": 3,
"y": 11,
"width": 3,
"height": 7
}
}
]
}
]
}