Senior Data Engineer

Name: Senior Data Engineer
Author: ovachiever

ovachiever/droid-tings

50 installs
51 repo stars
Updated November 25, 2025
ovachiever/droid-tings

This is a copy of senior-data-engineer by davila7 - installs and ranking accrue to the original listing.

Helps with ai & agent building tasks during AI-assisted development.

About

senior-data-engineer is a Claude Code skill for ai & agent building. It helps solo builders move faster with AI-assisted coding.

senior-data-engineer
AI & Agent Building
AI-coding skill

Senior Data Engineer by the numbers

50 all-time installs (skills.sh)
Data as of Jul 27, 2026 (Skillselion catalog sync)

npx skills add https://github.com/ovachiever/droid-tings --skill senior-data-engineer

Add your badge

Show developers this skill is listed on Skillselion. Paste this into your README.

[![Listed on Skillselion](https://skillselion.com/badge/skills/ovachiever/droid-tings/senior-data-engineer.svg)](https://skillselion.com/skills/ovachiever/droid-tings/senior-data-engineer)

Installs	50
repo stars	★ 51
Last updated	November 25, 2025
Repository	ovachiever/droid-tings ↗

What it does

Helps with ai & agent building tasks during AI-assisted development.

Files

SKILL.mdMarkdownGitHub ↗

Senior Data Engineer

World-class senior data engineer skill for production-grade AI/ML/Data systems.

Quick Start

Main Capabilities

# Core Tool 1
python scripts/pipeline_orchestrator.py --input data/ --output results/

# Core Tool 2  
python scripts/data_quality_validator.py --target project/ --analyze

# Core Tool 3
python scripts/etl_performance_optimizer.py --config config.yaml --deploy

Core Expertise

This skill covers world-class capabilities in:

Advanced production patterns and architectures
Scalable system design and implementation
Performance optimization at scale
MLOps and DataOps best practices
Real-time processing and inference
Distributed computing frameworks
Model deployment and monitoring
Security and compliance
Cost optimization
Team leadership and mentoring

Tech Stack

Languages: Python, SQL, R, Scala, Go ML Frameworks: PyTorch, TensorFlow, Scikit-learn, XGBoost Data Tools: Spark, Airflow, dbt, Kafka, Databricks LLM Frameworks: LangChain, LlamaIndex, DSPy Deployment: Docker, Kubernetes, AWS/GCP/Azure Monitoring: MLflow, Weights & Biases, Prometheus Databases: PostgreSQL, BigQuery, Snowflake, Pinecone

Reference Documentation

1. Data Pipeline Architecture

Comprehensive guide available in references/data_pipeline_architecture.md covering:

Advanced patterns and best practices
Production implementation strategies
Performance optimization techniques
Scalability considerations
Security and compliance
Real-world case studies

2. Data Modeling Patterns

Complete workflow documentation in references/data_modeling_patterns.md including:

Step-by-step processes
Architecture design patterns
Tool integration guides
Performance tuning strategies
Troubleshooting procedures

3. Dataops Best Practices

Technical reference guide in references/dataops_best_practices.md with:

System design principles
Implementation examples
Configuration best practices
Deployment strategies
Monitoring and observability

Production Patterns

Pattern 1: Scalable Data Processing

Enterprise-scale data processing with distributed computing:

Horizontal scaling architecture
Fault-tolerant design
Real-time and batch processing
Data quality validation
Performance monitoring

Pattern 2: ML Model Deployment

Production ML system with high availability:

Model serving with low latency
A/B testing infrastructure
Feature store integration
Model monitoring and drift detection
Automated retraining pipelines

Pattern 3: Real-Time Inference

High-throughput inference system:

Batching and caching strategies
Load balancing
Auto-scaling
Latency optimization
Cost optimization

Best Practices

Development

Test-driven development
Code reviews and pair programming
Documentation as code
Version control everything
Continuous integration

Production

Monitor everything critical
Automate deployments
Feature flags for releases
Canary deployments
Comprehensive logging

Team Leadership

Mentor junior engineers
Drive technical decisions
Establish coding standards
Foster learning culture
Cross-functional collaboration

Performance Targets

Latency:

P50: < 50ms
P95: < 100ms
P99: < 200ms

Throughput:

Requests/second: > 1000
Concurrent users: > 10,000

Availability:

Uptime: 99.9%
Error rate: < 0.1%

Security & Compliance

Authentication & authorization
Data encryption (at rest & in transit)
PII handling and anonymization
GDPR/CCPA compliance
Regular security audits
Vulnerability management

Common Commands

# Development
python -m pytest tests/ -v --cov
python -m black src/
python -m pylint src/

# Training
python scripts/train.py --config prod.yaml
python scripts/evaluate.py --model best.pth

# Deployment
docker build -t service:v1 .
kubectl apply -f k8s/
helm upgrade service ./charts/

# Monitoring
kubectl logs -f deployment/service
python scripts/health_check.py

Resources

Advanced Patterns: references/data_pipeline_architecture.md
Implementation Guide: references/data_modeling_patterns.md
Technical Reference: references/dataops_best_practices.md
Automation Scripts: scripts/ directory

Senior-Level Responsibilities

As a world-class senior professional:

1. Technical Leadership

Drive architectural decisions
Mentor team members
Establish best practices
Ensure code quality

2. Strategic Thinking

Align with business goals
Evaluate trade-offs
Plan for scale
Manage technical debt

3. Collaboration

Work across teams
Communicate effectively
Build consensus
Share knowledge

4. Innovation

Stay current with research
Experiment with new approaches
Contribute to community
Drive continuous improvement

5. Production Excellence

Ensure high availability
Monitor proactively
Optimize performance
Respond to incidents

#!/usr/bin/env python3
"""
Data Quality Validator
Production-grade tool for senior data engineer
"""

import os
import sys
import json
import logging
import argparse
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class DataQualityValidator:
    """Production-grade data quality validator"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.results = {
            'status': 'initialized',
            'start_time': datetime.now().isoformat(),
            'processed_items': 0
        }
        logger.info(f"Initialized {self.__class__.__name__}")
    
    def validate_config(self) -> bool:
        """Validate configuration"""
        logger.info("Validating configuration...")
        # Add validation logic
        logger.info("Configuration validated")
        return True
    
    def process(self) -> Dict:
        """Main processing logic"""
        logger.info("Starting processing...")
        
        try:
            self.validate_config()
            
            # Main processing
            result = self._execute()
            
            self.results['status'] = 'completed'
            self.results['end_time'] = datetime.now().isoformat()
            
            logger.info("Processing completed successfully")
            return self.results
            
        except Exception as e:
            self.results['status'] = 'failed'
            self.results['error'] = str(e)
            logger.error(f"Processing failed: {e}")
            raise
    
    def _execute(self) -> Dict:
        """Execute main logic"""
        # Implementation here
        return {'success': True}

def main():
    """Main entry point"""
    parser = argparse.ArgumentParser(
        description="Data Quality Validator"
    )
    parser.add_argument('--input', '-i', required=True, help='Input path')
    parser.add_argument('--output', '-o', required=True, help='Output path')
    parser.add_argument('--config', '-c', help='Configuration file')
    parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    
    try:
        config = {
            'input': args.input,
            'output': args.output
        }
        
        processor = DataQualityValidator(config)
        results = processor.process()
        
        print(json.dumps(results, indent=2))
        sys.exit(0)
        
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Etl Performance Optimizer
Production-grade tool for senior data engineer
"""

import os
import sys
import json
import logging
import argparse
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class EtlPerformanceOptimizer:
    """Production-grade etl performance optimizer"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.results = {
            'status': 'initialized',
            'start_time': datetime.now().isoformat(),
            'processed_items': 0
        }
        logger.info(f"Initialized {self.__class__.__name__}")
    
    def validate_config(self) -> bool:
        """Validate configuration"""
        logger.info("Validating configuration...")
        # Add validation logic
        logger.info("Configuration validated")
        return True
    
    def process(self) -> Dict:
        """Main processing logic"""
        logger.info("Starting processing...")
        
        try:
            self.validate_config()
            
            # Main processing
            result = self._execute()
            
            self.results['status'] = 'completed'
            self.results['end_time'] = datetime.now().isoformat()
            
            logger.info("Processing completed successfully")
            return self.results
            
        except Exception as e:
            self.results['status'] = 'failed'
            self.results['error'] = str(e)
            logger.error(f"Processing failed: {e}")
            raise
    
    def _execute(self) -> Dict:
        """Execute main logic"""
        # Implementation here
        return {'success': True}

def main():
    """Main entry point"""
    parser = argparse.ArgumentParser(
        description="Etl Performance Optimizer"
    )
    parser.add_argument('--input', '-i', required=True, help='Input path')
    parser.add_argument('--output', '-o', required=True, help='Output path')
    parser.add_argument('--config', '-c', help='Configuration file')
    parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    
    try:
        config = {
            'input': args.input,
            'output': args.output
        }
        
        processor = EtlPerformanceOptimizer(config)
        results = processor.process()
        
        print(json.dumps(results, indent=2))
        sys.exit(0)
        
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Pipeline Orchestrator
Production-grade tool for senior data engineer
"""

import os
import sys
import json
import logging
import argparse
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class PipelineOrchestrator:
    """Production-grade pipeline orchestrator"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.results = {
            'status': 'initialized',
            'start_time': datetime.now().isoformat(),
            'processed_items': 0
        }
        logger.info(f"Initialized {self.__class__.__name__}")
    
    def validate_config(self) -> bool:
        """Validate configuration"""
        logger.info("Validating configuration...")
        # Add validation logic
        logger.info("Configuration validated")
        return True
    
    def process(self) -> Dict:
        """Main processing logic"""
        logger.info("Starting processing...")
        
        try:
            self.validate_config()
            
            # Main processing
            result = self._execute()
            
            self.results['status'] = 'completed'
            self.results['end_time'] = datetime.now().isoformat()
            
            logger.info("Processing completed successfully")
            return self.results
            
        except Exception as e:
            self.results['status'] = 'failed'
            self.results['error'] = str(e)
            logger.error(f"Processing failed: {e}")
            raise
    
    def _execute(self) -> Dict:
        """Execute main logic"""
        # Implementation here
        return {'success': True}

def main():
    """Main entry point"""
    parser = argparse.ArgumentParser(
        description="Pipeline Orchestrator"
    )
    parser.add_argument('--input', '-i', required=True, help='Input path')
    parser.add_argument('--output', '-o', required=True, help='Output path')
    parser.add_argument('--config', '-c', help='Configuration file')
    parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    
    try:
        config = {
            'input': args.input,
            'output': args.output
        }
        
        processor = PipelineOrchestrator(config)
        results = processor.process()
        
        print(json.dumps(results, indent=2))
        sys.exit(0)
        
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == '__main__':
    main()

Related skills

Setup Matt Pocock SkillsScaffold the per-repo configuration that Matt Pocock’s engineering agent skills rely on so they understand the issue tracker, triage labels, and domain documentation la495k196k

Lark AppsConnect Claude, Cursor or custom agents directly to Lark (Feishu) for messaging, document automation, approval workflows and enterprise data access.402k

CavemanSlash token usage by roughly 75% while keeping every technical detail intact when working with Claude Code, Cursor or similar agents.391k94.7k

Lark Skill MakerQuickly turn any Lark/Feishu OpenAPI call or multi-step workflow into a reusable agent skill with its own SKILL.md.386k16k

Running Claude Code Via Litellm CopilotRun Claude Code at a fraction of the cost by routing requests through LiteLLM to the GitHub Copilot Chat API.270k72

Codex PetGenerate a complete Codex Pet spritesheet and metadata from one reference image without needing an OpenAI key or Codex Pro.246k10

AI & Agent Buildingagents

About

Senior Data Engineer by the numbers

Add your badge

What it does

Files

Senior Data Engineer

Quick Start

Main Capabilities

Core Expertise

Tech Stack

Reference Documentation

1. Data Pipeline Architecture

2. Data Modeling Patterns

3. Dataops Best Practices

Production Patterns

Pattern 1: Scalable Data Processing

Pattern 2: ML Model Deployment

Pattern 3: Real-Time Inference

Best Practices

Development

Production

Team Leadership

Performance Targets

Security & Compliance

Common Commands

Resources

Senior-Level Responsibilities

Data Modeling Patterns

Overview

Core Principles

Production-First Design

Performance by Design

Security & Privacy

Advanced Patterns

Pattern 1: Distributed Processing

Pattern 2: Real-Time Systems

Pattern 3: ML at Scale

Best Practices

Code Quality

Performance

Reliability

Tools & Technologies

Further Reading

Data Pipeline Architecture

Overview

Core Principles

Production-First Design

Performance by Design

Security & Privacy

Advanced Patterns

Pattern 1: Distributed Processing

Pattern 2: Real-Time Systems

Pattern 3: ML at Scale

Best Practices

Code Quality

Performance

Reliability

Tools & Technologies

Further Reading

Dataops Best Practices

Overview

Core Principles

Production-First Design

Performance by Design

Security & Privacy

Advanced Patterns

Pattern 1: Distributed Processing

Pattern 2: Real-Time Systems

Pattern 3: ML at Scale

Best Practices

Code Quality

Performance

Reliability

Tools & Technologies

Further Reading

Related skills

This week in AI coding