# Data Lake > "Provides Data lake architecture and management for trading data storage" - Skill: `paulpas/data-lake` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/data-lake` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/data-lake/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/data-lake --- **Role:** Provide scalable, cost-effective storage for trading data with efficient query capabilities **Philosophy:** Data is an asset; storage must balance cost, accessibility, and retention policies ## Key Principles 1. **Partitioned Storage**: Organize data by symbol, date, and time for efficient access 2. **Columnar Format**: Use Parquet for compressed, columnar storage 3. **Hot/Warm/Cold Tiers**: Automatically move data based on access patterns 4. **Data Lifecycle Policies**: Define retention and archival rules 5. **Metadata Catalog**: Track schema, size, and access patterns ## Implementation Guidelines ### Structure - Core logic: data_lake/lake_manager.py - Storage adapter: data_lake/storage.py - Tests: tests/test_data_lake.py ### Patterns to Follow - Use partitioned directory structure - Implement batch writes for efficiency - Support time-travel queries - Track storage metrics ## Adherence Checklist Before completing your task, verify: - [ ] Data is partitioned for efficient queries - [ ] Columnar compression reduces storage costs - [ ] Lifecycle policies are enforced - [ ] Metadata catalog is updated on writes - [ ] Access patterns are monitored Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory. ## Python Implementation ```python import os import time from typing import Dict, List, Optional, Any, Iterable from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum import json import logging class StorageTier(Enum): HOT = "hot" # Fast access, SSD WARM = "warm" # Moderate access, HDD COLD = "cold" # Infrequent access, archive @dataclass class DataFile: """Represents a data file in the lake.""" path: str size_bytes: int row_count: int partition: Dict[str, str] created_at: float last_accessed: float tier: StorageTier @dataclass class DataLakeConfig: """Configuration for data lake.""" base_path: str partition_columns: List[str] = field(default_factory=lambda: ["symbol", "date", "hour"]) retention_days: int = 365 compression: str = "snappy" file_format: str = "parquet" class DataLake: """Manages data storage in the data lake.""" def __init__(self, config: DataLakeConfig): self.config = config self.files: Dict[str, DataFile] = {} self._metadata_path = os.path.join(config.base_path, "_metadata") self._initialize_metadata() def _initialize_metadata(self): """Initialize metadata storage.""" os.makedirs(self._metadata_path, exist_ok=True) self.metadata_file = os.path.join(self._metadata_path, "files.json") if os.path.exists(self.metadata_file): try: with open(self.metadata_file, 'r') as f: data = json.load(f) for path, file_data in data.items(): self.files[path] = DataFile(**file_data) except Exception as e: logging.warning(f"Failed to load metadata: {e}") def _save_metadata(self): """Save metadata to disk.""" data = { path: { "path": f.path, "size_bytes": f.size_bytes, "row_count": f.row_count, "partition": f.partition, "created_at": f.created_at, "last_accessed": f.last_accessed, "tier": f.tier.value } for path, f in self.files.items() } os.makedirs(self._metadata_path, exist_ok=True) with open(self.metadata_file, 'w') as f: json.dump(data, f, indent=2) def write_data( self, data: List[Dict[str, Any]], partition_values: Dict[str, str] ) -> str: """Write data to lake with partitioning.""" partition_path = self._build_partition_path(partition_values) filepath = os.path.join( self.config.base_path, partition_path, f"data_{int(time.time() * 1000)}.parquet" ) # In a real implementation, this would write to Parquet # For demonstration, just create the directory structure os.makedirs(os.path.dirname(filepath), exist_ok=True) file_info = DataFile( path=filepath, size_bytes=len(str(data)), # Approximate row_count=len(data), partition=partition_values, created_at=time.time(), last_accessed=time.time(), tier=StorageTier.HOT ) self.files[filepath] = file_info self._save_metadata() return filepath def _build_partition_path(self, partition_values: Dict[str, str]) -> str: """Build partition directory path.""" parts = [] for col in self.config.partition_columns: if col in partition_values: parts.append(f"{col}={partition_values[col]}") return "/".join(parts) def read_data( self, symbol: str, start_time: float, end_time: float ) -> List[Dict[str, Any]]: """Read data for a symbol and time range.""" # Find relevant files matching_files = [] for file in self.files.values(): if symbol == file.partition.get("symbol", symbol): # Check time range (simplified) if start_time <= file.last_accessed <= end_time: matching_files.append(file) # In a real implementation, this would read from storage return [] def query( self, symbol: Optional[str] = None, date_range: Optional[tuple] = None, limit: int = 1000 ) -> List[Dict[str, Any]]: """Query data lake with filters.""" results = [] for file in self.files.values(): # Apply filters if symbol and file.partition.get("symbol") != symbol: continue results.append({ "path": file.path, "partition": file.partition, "size_bytes": file.size_bytes, "row_count": file.row_count }) if len(results) >= limit: break return results def get_file(self, path: str) -> Optional[DataFile]: """Get file metadata.""" return self.files.get(path) def list_files( self, symbol: Optional[str] = None, tier: Optional[StorageTier] = None ) -> List[DataFile]: """List files with optional filters.""" results = [] for file in self.files.values(): if symbol and file.partition.get("symbol") != symbol: continue if tier and file.tier != tier: continue results.append(file) return results def delete_file(self, path: str) -> bool: """Delete a file.""" if path in self.files: del self.files[path] self._save_metadata() return True return False def get_stats(self) -> Dict[str, Any]: """Get data lake statistics.""" total_size = sum(f.size_bytes for f in self.files.values()) total_rows = sum(f.row_count for f in self.files.values()) tier_counts = {} for f in self.files.values(): tier = f.tier.value tier_counts[tier] = tier_counts.get(tier, 0) + 1 return { "total_files": len(self.files), "total_size_bytes": total_size, "total_rows": total_rows, "files_by_tier": tier_counts, "avg_file_size": total_size / len(self.files) if self.files else 0 } class DataLifecycleManager: """Manages data lifecycle policies.""" def __init__(self, lake: DataLake): self.lake = lake self.policies: Dict[str, Dict] = {} self._initialize_policies() def _initialize_policies(self): """Initialize lifecycle policies.""" self.policies = { "hot": {"max_age_days": 7, "tier": StorageTier.HOT}, "warm": {"max_age_days": 30, "tier": StorageTier.WARM}, "cold": {"max_age_days": 365, "tier": StorageTier.COLD} } def set_policy(self, symbol: str, hot_days: int, warm_days: int): """Set lifecycle policy for a symbol.""" self.policies[symbol] = { "hot_days": hot_days, "warm_days": warm_days } def get_tier_for_file(self, file: DataFile, symbol: str) -> StorageTier: """Determine tier for a file based on age.""" if symbol in self.policies: policy = self.policies[symbol] age_days = (time.time() - file.created_at) / 86400 if age_days < policy["hot_days"]: return StorageTier.HOT elif age_days < policy["warm_days"]: return StorageTier.WARM else: return StorageTier.COLD # Default policy age_days = (time.time() - file.created_at) / 86400 if age_days < 7: return StorageTier.HOT elif age_days < 30: return StorageTier.WARM return StorageTier.COLD def enforce_policies(self) -> List[Dict]: """Enforce lifecycle policies and return tier changes.""" changes = [] for file in self.lake.files.values(): symbol = file.partition.get("symbol") if not symbol: continue new_tier = self.get_tier_for_file(file, symbol) if new_tier != file.tier: old_tier = file.tier file.tier = new_tier changes.append({ "path": file.path, "from_tier": old_tier.value, "to_tier": new_tier.value, "age_days": (time.time() - file.created_at) / 86400 }) self.lake._save_metadata() return changes class DataCatalog: """Catalog of data lake contents.""" def __init__(self, lake: DataLake): self.lake = lake self.schema_registry: Dict[str, Dict] = {} def register_schema(self, symbol: str, schema: Dict[str, str]): """Register schema for a symbol.""" self.schema_registry[symbol] = { "schema": schema, "updated_at": time.time() } def get_schema(self, symbol: str) -> Optional[Dict[str, str]]: """Get schema for a symbol.""" if symbol in self.schema_registry: return self.schema_registry[symbol]["schema"] return None def search(self, keyword: str) -> List[Dict]: """Search catalog by keyword.""" results = [] for file in self.lake.files.values(): symbol = file.partition.get("symbol", "") if keyword.lower() in symbol.lower() or keyword in file.path.lower(): results.append({ "symbol": symbol, "path": file.path, "partition": file.partition, "size_bytes": file.size_bytes, "created_at": file.created_at }) return results ``` --- --- ### Pattern 2: Parquet Data Lake with Schema Evolution and Partitioning ```python from __future__ import annotations import logging from datetime import datetime, timezone from pathlib import Path from typing import Optional import pandas as pd logger = logging.getLogger(__name__) class DataLakeWriter: """Writes trading data to a partitioned Parquet data lake with schema management.""" def __init__(self, base_path: str = "/data/lake/trading"): self.base_path = Path(base_path) self._schema_version: dict[str, int] = {} def write_candles( self, df: pd.DataFrame, symbol: str, timeframe: str = "1h", schema_version: int = 1, ) -> dict: """Write OHLCV data to the data lake in partitioned Parquet format. Partitions by date (year/month/day) for efficient time-range queries. Args: df: DataFrame with columns: timestamp, open, high, low, close, volume. symbol: Trading pair identifier. timeframe: Data granularity used for partition naming. schema_version: Current schema version number. Returns: Summary dict with file paths written and row counts. """ df = df.copy() df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True) df["symbol"] = symbol df["timeframe"] = timeframe # Create date partitions df["date"] = df["timestamp"].dt.date result = {"files_written": [], "total_rows": 0} for date_val, group in df.groupby("date"): partition_path = self.base_path / symbol / str(timeframe) / f"year={date_val.year}" / f"month={date_val.month:02d}" / f"day={date_val.day:02d}" partition_path.mkdir(parents=True, exist_ok=True) file_name = partition_path / f"candles_{symbol}_{timeframe}_{date_val.isoformat()}.parquet" group[[ "timestamp", "open", "high", "low", "close", "volume", "symbol", "timeframe", "date", ]].to_parquet(file_name, engine="pyarrow", index=False) result["files_written"].append(str(file_name)) result["total_rows"] += len(group) self._schema_version[symbol] = schema_version logger.info("Wrote %d candle rows for %s to data lake (v%d)", result["total_rows"], symbol, schema_version) return result def read_range( self, symbol: str, start: datetime, end: datetime, timeframe: str = "1h", ) -> pd.DataFrame: """Read candle data from the data lake for a specific time range. Uses partition pruning to scan only relevant date directories. """ files = [] current = start.date() end_date = end.date() while current <= end_date: search_path = ( self.base_path / symbol / timeframe / f"year={current.year}" / f"month={current.month:02d}" / f"day={current.day:02d}" ) if search_path.exists(): for pf in search_path.glob(f"candles_{symbol}_{timeframe}_*.parquet"): files.append(str(pf)) current += __import__("datetime").timedelta(days=1) if not files: logger.warning("No data found for %s in range %s → %s", symbol, start, end) return pd.DataFrame() df = pd.concat([pd.read_parquet(f) for f in files], ignore_index=True) mask = (df["timestamp"] >= start) & (df["timestamp"] < end) return df.loc[mask].sort_values("timestamp").reset_index(drop=True) ``` ## Constraints ### MUST DO - Validate all incoming data against schema constraints (type, range, nullability) before processing or storage - Implement idempotent operations: re-processing the same data must produce identical results - Track data lineage and provenance with timestamps, source identifiers, and transformation history for every record - Handle out-of-order data by implementing a watermark-based ordering mechanism with configurable tolerance window - Log data quality metrics (completeness, freshness, accuracy) per source with automatic alerting on degradation ### MUST NOT DO - Do not silently drop records that fail validation — log them to a quarantine table for review - Avoid concatenating strings for timestamp comparison; use proper datetime/timedelta objects - Never assume data arrives in chronological order from any external feed without explicit ordering guarantees - Do not store raw and processed data in the same table without clear partitioning or separation strategy - Avoid blocking on slow data sources — implement async prefetch with timeout-based fallback to cached data ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [What is a Data Lake?](https://aws.amazon.com/big-data/datalakes-and-analytics/what-is-a-data-lake/) - [AWS Data Lake Architecture](https://aws.amazon.com/solutions/guidance/an-end-to-end-data-lake/) - [Delta Lake for Financial Data](https://delta.io/) - [Data Lake vs Data Warehouse Comparison](https://www.investopedia.com/terms/d/data-lake.asp) - [Building Scalable Data Lakes](https://aws.amazon.com/big-data/datalakes-and-analytics/batch-processing-tools/)