data-engineering-study-material

Data Engineering Study Material Skill by ara.so — Data Skills collection. Overview This project is a comprehensive study guide and reference repository for data engineering concepts, tools, and practices. It serves as a centralized resource for learning core data engineering principles, understanding modern data stack components, and preparing for data engineering roles. The repository covers: - Data engineering fundamentals and architecture patterns - ETL/ELT pipeline design and implementation - Data warehousing and lake architectures - Streaming and batch processing frameworks - Cloud data…

),\n 'amount': lambda col: col.expect_column_values_to_be_between(min_value=0, max_value=1000000),\n 'created_at': lambda col: col.expect_column_values_to_not_be_null(),\n 'status': lambda col: col.expect_column_values_to_be_in_set(['active', 'inactive', 'pending'])\n }\n \n # Run validations\n results = []\n for column, expectation_func in expectations.items():\n if column in ge_df.columns:\n result = expectation_func(ge_df[column])\n results.append(result)\n \n # Check if all validations passed\n all_passed = all(r.success for r in results)\n \n return all_passed, results\n```\n\n### Apache Airflow DAG Example\n\n```python\nfrom airflow import DAG\nfrom airflow.operators.python import PythonOperator\nfrom airflow.providers.postgres.operators.postgres import PostgresOperator\nfrom airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator\nfrom datetime import datetime, timedelta\n\ndefault_args = {\n 'owner': 'data-engineering',\n 'depends_on_past': False,\n 'start_date': datetime(2024, 1, 1),\n 'email_on_failure': True,\n 'email_on_retry': False,\n 'retries': 2,\n 'retry_delay': timedelta(minutes=5)\n}\n\ndef extract_from_api(**context):\n \"\"\"Extract data from external API\"\"\"\n import requests\n import json\n \n response = requests.get('https://api.example.com/data')\n data = response.json()\n \n # Save to S3\n s3_path = f\"s3://my-bucket/raw/{context['ds']}/data.json\"\n # Upload logic here\n \n return s3_path\n\ndef transform_data(**context):\n \"\"\"Transform extracted data\"\"\"\n import pandas as pd\n \n # Read from S3\n s3_path = context['ti'].xcom_pull(task_ids='extract_task')\n df = pd.read_json(s3_path)\n \n # Transformations\n df_transformed = df.drop_duplicates()\n df_transformed['load_date'] = context['ds']\n \n # Write back to S3\n output_path = f\"s3://my-bucket/processed/{context['ds']}/data.parquet\"\n df_transformed.to_parquet(output_path)\n \n return output_path\n\nwith DAG(\n 'daily_etl_pipeline',\n default_args=default_args,\n description='Daily ETL pipeline for data processing',\n schedule_interval='0 2 * * *', # Run at 2 AM daily\n catchup=False,\n tags=['etl', 'daily']\n) as dag:\n \n extract_task = PythonOperator(\n task_id='extract_task',\n python_callable=extract_from_api,\n provide_context=True\n )\n \n transform_task = PythonOperator(\n task_id='transform_task',\n python_callable=transform_data,\n provide_context=True\n )\n \n load_task = S3ToRedshiftOperator(\n task_id='load_to_redshift',\n s3_bucket='my-bucket',\n s3_key='processed/{{ ds }}/data.parquet',\n schema='analytics',\n table='fact_table',\n copy_options=['FORMAT AS PARQUET']\n )\n \n data_quality_check = PostgresOperator(\n task_id='quality_check',\n postgres_conn_id='redshift_conn',\n sql=\"\"\"\n SELECT COUNT(*) as row_count \n FROM analytics.fact_table \n WHERE load_date = '{{ ds }}';\n \"\"\"\n )\n \n extract_task >> transform_task >> load_task >> data_quality_check\n```\n\n### Spark Batch Processing Example\n\n```python\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col, when, sum, avg, count, to_date\nfrom pyspark.sql.window import Window\n\ndef process_batch_data():\n \"\"\"Process large-scale batch data with Apache Spark\"\"\"\n \n # Initialize Spark session\n spark = SparkSession.builder \\\n .appName(\"BatchDataProcessing\") \\\n .config(\"spark.sql.adaptive.enabled\", \"true\") \\\n .getOrCreate()\n \n # Read data from data lake\n df = spark.read \\\n .format(\"parquet\") \\\n .load(\"s3a://data-lake/raw/transactions/\")\n \n # Transformations\n df_transformed = df \\\n .withColumn(\"transaction_date\", to_date(col(\"timestamp\"))) \\\n .filter(col(\"amount\") > 0) \\\n .dropDuplicates([\"transaction_id\"])\n \n # Aggregations\n daily_summary = df_transformed.groupBy(\"transaction_date\", \"category\") \\\n .agg(\n count(\"transaction_id\").alias(\"transaction_count\"),\n sum(\"amount\").alias(\"total_amount\"),\n avg(\"amount\").alias(\"avg_amount\")\n )\n \n # Window functions for ranking\n window_spec = Window.partitionBy(\"transaction_date\").orderBy(col(\"total_amount\").desc())\n \n ranked_summary = daily_summary \\\n .withColumn(\"rank\", dense_rank().over(window_spec)) \\\n .filter(col(\"rank\") \u003c= 10)\n \n # Write to data warehouse\n ranked_summary.write \\\n .format(\"parquet\") \\\n .mode(\"overwrite\") \\\n .partitionBy(\"transaction_date\") \\\n .save(\"s3a://data-warehouse/analytics/daily_summary/\")\n \n spark.stop()\n\nif __name__ == \"__main__\":\n process_batch_data()\n```\n\n### Streaming Pipeline Example (Kafka + Spark Structured Streaming)\n\n```python\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.functions import from_json, col, window\nfrom pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType\n\ndef create_streaming_pipeline():\n \"\"\"Real-time data processing with Spark Structured Streaming\"\"\"\n \n spark = SparkSession.builder \\\n .appName(\"RealTimeStreaming\") \\\n .getOrCreate()\n \n # Define schema for incoming data\n schema = StructType([\n StructField(\"event_id\", StringType(), True),\n StructField(\"user_id\", StringType(), True),\n StructField(\"event_type\", StringType(), True),\n StructField(\"amount\", DoubleType(), True),\n StructField(\"timestamp\", TimestampType(), True)\n ])\n \n # Read from Kafka\n df_stream = spark.readStream \\\n .format(\"kafka\") \\\n .option(\"kafka.bootstrap.servers\", \"localhost:9092\") \\\n .option(\"subscribe\", \"events\") \\\n .option(\"startingOffsets\", \"latest\") \\\n .load()\n \n # Parse JSON data\n df_parsed = df_stream.select(\n from_json(col(\"value\").cast(\"string\"), schema).alias(\"data\")\n ).select(\"data.*\")\n \n # Windowed aggregations\n df_windowed = df_parsed \\\n .withWatermark(\"timestamp\", \"10 minutes\") \\\n .groupBy(\n window(col(\"timestamp\"), \"5 minutes\", \"1 minute\"),\n col(\"event_type\")\n ) \\\n .agg(\n count(\"event_id\").alias(\"event_count\"),\n sum(\"amount\").alias(\"total_amount\")\n )\n \n # Write to sink\n query = df_windowed.writeStream \\\n .outputMode(\"append\") \\\n .format(\"parquet\") \\\n .option(\"path\", \"s3a://streaming-output/events/\") \\\n .option(\"checkpointLocation\", \"s3a://checkpoints/events/\") \\\n .trigger(processingTime=\"1 minute\") \\\n .start()\n \n query.awaitTermination()\n\nif __name__ == \"__main__\":\n create_streaming_pipeline()\n```\n\n## Common Patterns\n\n### Incremental Data Loading\n\n```python\ndef incremental_load(table_name, watermark_column, last_watermark):\n \"\"\"Load only new/updated records since last run\"\"\"\n \n query = f\"\"\"\n SELECT *\n FROM {table_name}\n WHERE {watermark_column} > '{last_watermark}'\n ORDER BY {watermark_column}\n \"\"\"\n \n # Execute query and get new data\n new_data = pd.read_sql(query, source_conn)\n \n if not new_data.empty:\n # Get new watermark\n new_watermark = new_data[watermark_column].max()\n \n # Load to target\n new_data.to_sql('target_table', target_conn, if_exists='append', index=False)\n \n # Update watermark\n save_watermark(table_name, new_watermark)\n \n return len(new_data)\n```\n\n### SCD Type 2 Implementation\n\n```python\ndef scd_type2_merge(source_df, target_table, business_key, effective_date):\n \"\"\"Implement Slowly Changing Dimension Type 2\"\"\"\n \n from datetime import datetime\n \n # Read current dimension table\n current_df = pd.read_sql(f\"SELECT * FROM {target_table} WHERE is_current = 1\", conn)\n \n # Identify changes\n merged = source_df.merge(\n current_df,\n on=business_key,\n how='left',\n suffixes=('_new', '_old')\n )\n \n # Records that changed\n changed = merged[\n (merged.apply(lambda row: row_has_changes(row), axis=1))\n ]\n \n # Expire old records\n if not changed.empty:\n expire_query = f\"\"\"\n UPDATE {target_table}\n SET is_current = 0,\n end_date = '{effective_date}'\n WHERE {business_key} IN ({','.join(map(str, changed[business_key].tolist()))})\n AND is_current = 1\n \"\"\"\n conn.execute(expire_query)\n \n # Insert new versions\n new_records = source_df[source_df[business_key].isin(changed[business_key])]\n new_records['start_date'] = effective_date\n new_records['end_date'] = '9999-12-31'\n new_records['is_current'] = 1\n \n new_records.to_sql(target_table, conn, if_exists='append', index=False)\n```\n\n## Configuration\n\n### Database Connection Configuration\n\n```python\n# config.py\nimport os\n\nDATABASE_CONFIG = {\n 'source': {\n 'host': os.getenv('SOURCE_DB_HOST'),\n 'port': os.getenv('SOURCE_DB_PORT', 5432),\n 'database': os.getenv('SOURCE_DB_NAME'),\n 'user': os.getenv('SOURCE_DB_USER'),\n 'password': os.getenv('SOURCE_DB_PASSWORD')\n },\n 'warehouse': {\n 'host': os.getenv('WAREHOUSE_HOST'),\n 'port': os.getenv('WAREHOUSE_PORT', 5439),\n 'database': os.getenv('WAREHOUSE_DB'),\n 'user': os.getenv('WAREHOUSE_USER'),\n 'password': os.getenv('WAREHOUSE_PASSWORD')\n }\n}\n\nSPARK_CONFIG = {\n 'spark.executor.memory': '4g',\n 'spark.driver.memory': '2g',\n 'spark.sql.adaptive.enabled': 'true',\n 'spark.sql.adaptive.coalescePartitions.enabled': 'true'\n}\n\nAIRFLOW_CONFIG = {\n 'concurrency': 16,\n 'max_active_runs': 3,\n 'dagbag_import_timeout': 30\n}\n```\n\n## Troubleshooting\n\n### Common Issues and Solutions\n\n**Issue: Out of Memory in Spark Jobs**\n```python\n# Solution: Optimize memory usage\nspark = SparkSession.builder \\\n .config(\"spark.executor.memory\", \"8g\") \\\n .config(\"spark.driver.memory\", \"4g\") \\\n .config(\"spark.sql.shuffle.partitions\", \"200\") \\\n .config(\"spark.default.parallelism\", \"200\") \\\n .getOrCreate()\n\n# Use broadcast joins for small tables\nfrom pyspark.sql.functions import broadcast\nresult = large_df.join(broadcast(small_df), \"key\")\n```\n\n**Issue: Slow Incremental Loads**\n```python\n# Solution: Use partition pruning and indexing\n# Add indexes on watermark columns\n# Partition target tables by date\n\n# Use partition pruning in Spark\ndf = spark.read.parquet(\"s3://data/table/\") \\\n .where(f\"partition_date >= '{start_date}'\")\n```\n\n**Issue: Data Quality Failures**\n```python\n# Solution: Implement comprehensive validation\ndef validate_and_quarantine(df, rules):\n \"\"\"Separate valid and invalid records\"\"\"\n \n valid_df = df\n invalid_records = []\n \n for rule_name, rule_func in rules.items():\n mask = rule_func(valid_df)\n invalid = valid_df[~mask].copy()\n invalid['failed_rule'] = rule_name\n invalid_records.append(invalid)\n valid_df = valid_df[mask]\n \n # Save invalid records for review\n if invalid_records:\n pd.concat(invalid_records).to_sql(\n 'data_quality_quarantine',\n conn,\n if_exists='append'\n )\n \n return valid_df\n```\n\n## Best Practices\n\n1. **Idempotency**: Ensure pipelines can be re-run safely\n2. **Monitoring**: Implement comprehensive logging and alerting\n3. **Data Quality**: Validate data at every stage\n4. **Partitioning**: Use appropriate partitioning strategies for performance\n5. **Documentation**: Document data lineage and transformations\n6. **Version Control**: Track schema changes and pipeline versions\n7. **Testing**: Test pipelines with sample data before production\n8. **Security**: Use IAM roles, encryption, and secure credential management\n\n## Interview Preparation\n\nCommon data engineering interview topics covered:\n- SQL optimization and query tuning\n- Distributed systems concepts\n- Data modeling (star schema, snowflake, Data Vault)\n- ETL vs ELT trade-offs\n- CAP theorem and consistency models\n- Data quality frameworks\n- Cloud platform services (S3, Redshift, BigQuery, Databricks)\n- Orchestration tools (Airflow, Prefect, Dagster)\n- Streaming architectures (Kafka, Kinesis, Pub/Sub)","attachment_filenames":[],"attachments":[],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Data Engineering Study Material","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Skill by ","type":"text"},{"text":"ara.so","type":"text","marks":[{"type":"link","attrs":{"href":"https://ara.so","title":null}}]},{"text":" — Data Skills collection.","type":"text"}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Overview","type":"text"}]},{"type":"paragraph","content":[{"text":"This project is a comprehensive study guide and reference repository for data engineering concepts, tools, and practices. It serves as a centralized resource for learning core data engineering principles, understanding modern data stack components, and preparing for data engineering roles.","type":"text"}]},{"type":"paragraph","content":[{"text":"The repository covers:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data engineering fundamentals and architecture patterns","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ETL/ELT pipeline design and implementation","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data warehousing and lake architectures","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Streaming and batch processing frameworks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Cloud data platforms (AWS, GCP, Azure)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data quality, governance, and observability","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Infrastructure as Code and orchestration tools","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Interview preparation and best practices","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Installation","type":"text"}]},{"type":"paragraph","content":[{"text":"This is a study material repository, not an installable package. Clone it to access the materials:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"git clone https://github.com/Ahmeduddin3403/data-engineering-study-material.git\ncd data-engineering-study-material","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Repository Structure","type":"text"}]},{"type":"paragraph","content":[{"text":"The materials are typically organized by topic area:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"data-engineering-study-material/\n├── fundamentals/ # Core concepts and principles\n├── tools/ # Tool-specific guides\n├── architectures/ # Design patterns and architectures\n├── pipelines/ # ETL/ELT examples\n├── cloud-platforms/ # Cloud-specific implementations\n├── streaming/ # Real-time processing\n├── batch-processing/ # Batch job patterns\n├── data-quality/ # Testing and validation\n├── orchestration/ # Workflow management\n├── interview-prep/ # Interview questions and answers\n└── projects/ # Hands-on project examples","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Core Data Engineering Concepts","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"ETL Pipeline Example (Python)","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import pandas as pd\nfrom sqlalchemy import create_engine\nimport logging\n\n# Configure logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\nclass ETLPipeline:\n \"\"\"Simple ETL pipeline for extracting, transforming, and loading data\"\"\"\n \n def __init__(self, source_path, target_conn_string):\n self.source_path = source_path\n self.engine = create_engine(target_conn_string)\n \n def extract(self):\n \"\"\"Extract data from source\"\"\"\n logger.info(f\"Extracting data from {self.source_path}\")\n df = pd.read_csv(self.source_path)\n logger.info(f\"Extracted {len(df)} rows\")\n return df\n \n def transform(self, df):\n \"\"\"Transform data: clean, deduplicate, enrich\"\"\"\n logger.info(\"Transforming data\")\n \n # Remove duplicates\n df = df.drop_duplicates()\n \n # Handle missing values\n df = df.fillna({\n 'numeric_column': 0,\n 'string_column': 'Unknown'\n })\n \n # Add derived columns\n df['created_date'] = pd.to_datetime(df['timestamp']).dt.date\n \n # Data validation\n df = df[df['amount'] > 0]\n \n logger.info(f\"Transformed to {len(df)} rows\")\n return df\n \n def load(self, df, table_name):\n \"\"\"Load data to target database\"\"\"\n logger.info(f\"Loading data to {table_name}\")\n df.to_sql(table_name, self.engine, if_exists='append', index=False)\n logger.info(\"Load complete\")\n \n def run(self, table_name):\n \"\"\"Execute full ETL pipeline\"\"\"\n try:\n df = self.extract()\n df_transformed = self.transform(df)\n self.load(df_transformed, table_name)\n logger.info(\"ETL pipeline completed successfully\")\n except Exception as e:\n logger.error(f\"ETL pipeline failed: {str(e)}\")\n raise\n\n# Usage\nif __name__ == \"__main__\":\n pipeline = ETLPipeline(\n source_path='data/raw/sales.csv',\n target_conn_string='postgresql://user:pass@localhost:5432/warehouse'\n )\n pipeline.run('sales_fact')","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Data Quality Checks","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import great_expectations as ge\n\ndef validate_data_quality(df):\n \"\"\"Implement data quality checks using Great Expectations\"\"\"\n \n # Convert pandas DataFrame to GE DataFrame\n ge_df = ge.from_pandas(df)\n \n # Define expectations\n expectations = {\n 'id': lambda col: col.expect_column_values_to_be_unique(),\n 'email': lambda col: col.expect_column_values_to_match_regex(r'^[\\w\\.-]+@[\\w\\.-]+\\.\\w+

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

),\n 'amount': lambda col: col.expect_column_values_to_be_between(min_value=0, max_value=1000000),\n 'created_at': lambda col: col.expect_column_values_to_not_be_null(),\n 'status': lambda col: col.expect_column_values_to_be_in_set(['active', 'inactive', 'pending'])\n }\n \n # Run validations\n results = []\n for column, expectation_func in expectations.items():\n if column in ge_df.columns:\n result = expectation_func(ge_df[column])\n results.append(result)\n \n # Check if all validations passed\n all_passed = all(r.success for r in results)\n \n return all_passed, results","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Apache Airflow DAG Example","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from airflow import DAG\nfrom airflow.operators.python import PythonOperator\nfrom airflow.providers.postgres.operators.postgres import PostgresOperator\nfrom airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator\nfrom datetime import datetime, timedelta\n\ndefault_args = {\n 'owner': 'data-engineering',\n 'depends_on_past': False,\n 'start_date': datetime(2024, 1, 1),\n 'email_on_failure': True,\n 'email_on_retry': False,\n 'retries': 2,\n 'retry_delay': timedelta(minutes=5)\n}\n\ndef extract_from_api(**context):\n \"\"\"Extract data from external API\"\"\"\n import requests\n import json\n \n response = requests.get('https://api.example.com/data')\n data = response.json()\n \n # Save to S3\n s3_path = f\"s3://my-bucket/raw/{context['ds']}/data.json\"\n # Upload logic here\n \n return s3_path\n\ndef transform_data(**context):\n \"\"\"Transform extracted data\"\"\"\n import pandas as pd\n \n # Read from S3\n s3_path = context['ti'].xcom_pull(task_ids='extract_task')\n df = pd.read_json(s3_path)\n \n # Transformations\n df_transformed = df.drop_duplicates()\n df_transformed['load_date'] = context['ds']\n \n # Write back to S3\n output_path = f\"s3://my-bucket/processed/{context['ds']}/data.parquet\"\n df_transformed.to_parquet(output_path)\n \n return output_path\n\nwith DAG(\n 'daily_etl_pipeline',\n default_args=default_args,\n description='Daily ETL pipeline for data processing',\n schedule_interval='0 2 * * *', # Run at 2 AM daily\n catchup=False,\n tags=['etl', 'daily']\n) as dag:\n \n extract_task = PythonOperator(\n task_id='extract_task',\n python_callable=extract_from_api,\n provide_context=True\n )\n \n transform_task = PythonOperator(\n task_id='transform_task',\n python_callable=transform_data,\n provide_context=True\n )\n \n load_task = S3ToRedshiftOperator(\n task_id='load_to_redshift',\n s3_bucket='my-bucket',\n s3_key='processed/{{ ds }}/data.parquet',\n schema='analytics',\n table='fact_table',\n copy_options=['FORMAT AS PARQUET']\n )\n \n data_quality_check = PostgresOperator(\n task_id='quality_check',\n postgres_conn_id='redshift_conn',\n sql=\"\"\"\n SELECT COUNT(*) as row_count \n FROM analytics.fact_table \n WHERE load_date = '{{ ds }}';\n \"\"\"\n )\n \n extract_task >> transform_task >> load_task >> data_quality_check","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Spark Batch Processing Example","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col, when, sum, avg, count, to_date\nfrom pyspark.sql.window import Window\n\ndef process_batch_data():\n \"\"\"Process large-scale batch data with Apache Spark\"\"\"\n \n # Initialize Spark session\n spark = SparkSession.builder \\\n .appName(\"BatchDataProcessing\") \\\n .config(\"spark.sql.adaptive.enabled\", \"true\") \\\n .getOrCreate()\n \n # Read data from data lake\n df = spark.read \\\n .format(\"parquet\") \\\n .load(\"s3a://data-lake/raw/transactions/\")\n \n # Transformations\n df_transformed = df \\\n .withColumn(\"transaction_date\", to_date(col(\"timestamp\"))) \\\n .filter(col(\"amount\") > 0) \\\n .dropDuplicates([\"transaction_id\"])\n \n # Aggregations\n daily_summary = df_transformed.groupBy(\"transaction_date\", \"category\") \\\n .agg(\n count(\"transaction_id\").alias(\"transaction_count\"),\n sum(\"amount\").alias(\"total_amount\"),\n avg(\"amount\").alias(\"avg_amount\")\n )\n \n # Window functions for ranking\n window_spec = Window.partitionBy(\"transaction_date\").orderBy(col(\"total_amount\").desc())\n \n ranked_summary = daily_summary \\\n .withColumn(\"rank\", dense_rank().over(window_spec)) \\\n .filter(col(\"rank\") \u003c= 10)\n \n # Write to data warehouse\n ranked_summary.write \\\n .format(\"parquet\") \\\n .mode(\"overwrite\") \\\n .partitionBy(\"transaction_date\") \\\n .save(\"s3a://data-warehouse/analytics/daily_summary/\")\n \n spark.stop()\n\nif __name__ == \"__main__\":\n process_batch_data()","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Streaming Pipeline Example (Kafka + Spark Structured Streaming)","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from pyspark.sql import SparkSession\nfrom pyspark.sql.functions import from_json, col, window\nfrom pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType\n\ndef create_streaming_pipeline():\n \"\"\"Real-time data processing with Spark Structured Streaming\"\"\"\n \n spark = SparkSession.builder \\\n .appName(\"RealTimeStreaming\") \\\n .getOrCreate()\n \n # Define schema for incoming data\n schema = StructType([\n StructField(\"event_id\", StringType(), True),\n StructField(\"user_id\", StringType(), True),\n StructField(\"event_type\", StringType(), True),\n StructField(\"amount\", DoubleType(), True),\n StructField(\"timestamp\", TimestampType(), True)\n ])\n \n # Read from Kafka\n df_stream = spark.readStream \\\n .format(\"kafka\") \\\n .option(\"kafka.bootstrap.servers\", \"localhost:9092\") \\\n .option(\"subscribe\", \"events\") \\\n .option(\"startingOffsets\", \"latest\") \\\n .load()\n \n # Parse JSON data\n df_parsed = df_stream.select(\n from_json(col(\"value\").cast(\"string\"), schema).alias(\"data\")\n ).select(\"data.*\")\n \n # Windowed aggregations\n df_windowed = df_parsed \\\n .withWatermark(\"timestamp\", \"10 minutes\") \\\n .groupBy(\n window(col(\"timestamp\"), \"5 minutes\", \"1 minute\"),\n col(\"event_type\")\n ) \\\n .agg(\n count(\"event_id\").alias(\"event_count\"),\n sum(\"amount\").alias(\"total_amount\")\n )\n \n # Write to sink\n query = df_windowed.writeStream \\\n .outputMode(\"append\") \\\n .format(\"parquet\") \\\n .option(\"path\", \"s3a://streaming-output/events/\") \\\n .option(\"checkpointLocation\", \"s3a://checkpoints/events/\") \\\n .trigger(processingTime=\"1 minute\") \\\n .start()\n \n query.awaitTermination()\n\nif __name__ == \"__main__\":\n create_streaming_pipeline()","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Common Patterns","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Incremental Data Loading","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"def incremental_load(table_name, watermark_column, last_watermark):\n \"\"\"Load only new/updated records since last run\"\"\"\n \n query = f\"\"\"\n SELECT *\n FROM {table_name}\n WHERE {watermark_column} > '{last_watermark}'\n ORDER BY {watermark_column}\n \"\"\"\n \n # Execute query and get new data\n new_data = pd.read_sql(query, source_conn)\n \n if not new_data.empty:\n # Get new watermark\n new_watermark = new_data[watermark_column].max()\n \n # Load to target\n new_data.to_sql('target_table', target_conn, if_exists='append', index=False)\n \n # Update watermark\n save_watermark(table_name, new_watermark)\n \n return len(new_data)","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"SCD Type 2 Implementation","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"def scd_type2_merge(source_df, target_table, business_key, effective_date):\n \"\"\"Implement Slowly Changing Dimension Type 2\"\"\"\n \n from datetime import datetime\n \n # Read current dimension table\n current_df = pd.read_sql(f\"SELECT * FROM {target_table} WHERE is_current = 1\", conn)\n \n # Identify changes\n merged = source_df.merge(\n current_df,\n on=business_key,\n how='left',\n suffixes=('_new', '_old')\n )\n \n # Records that changed\n changed = merged[\n (merged.apply(lambda row: row_has_changes(row), axis=1))\n ]\n \n # Expire old records\n if not changed.empty:\n expire_query = f\"\"\"\n UPDATE {target_table}\n SET is_current = 0,\n end_date = '{effective_date}'\n WHERE {business_key} IN ({','.join(map(str, changed[business_key].tolist()))})\n AND is_current = 1\n \"\"\"\n conn.execute(expire_query)\n \n # Insert new versions\n new_records = source_df[source_df[business_key].isin(changed[business_key])]\n new_records['start_date'] = effective_date\n new_records['end_date'] = '9999-12-31'\n new_records['is_current'] = 1\n \n new_records.to_sql(target_table, conn, if_exists='append', index=False)","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Configuration","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Database Connection Configuration","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# config.py\nimport os\n\nDATABASE_CONFIG = {\n 'source': {\n 'host': os.getenv('SOURCE_DB_HOST'),\n 'port': os.getenv('SOURCE_DB_PORT', 5432),\n 'database': os.getenv('SOURCE_DB_NAME'),\n 'user': os.getenv('SOURCE_DB_USER'),\n 'password': os.getenv('SOURCE_DB_PASSWORD')\n },\n 'warehouse': {\n 'host': os.getenv('WAREHOUSE_HOST'),\n 'port': os.getenv('WAREHOUSE_PORT', 5439),\n 'database': os.getenv('WAREHOUSE_DB'),\n 'user': os.getenv('WAREHOUSE_USER'),\n 'password': os.getenv('WAREHOUSE_PASSWORD')\n }\n}\n\nSPARK_CONFIG = {\n 'spark.executor.memory': '4g',\n 'spark.driver.memory': '2g',\n 'spark.sql.adaptive.enabled': 'true',\n 'spark.sql.adaptive.coalescePartitions.enabled': 'true'\n}\n\nAIRFLOW_CONFIG = {\n 'concurrency': 16,\n 'max_active_runs': 3,\n 'dagbag_import_timeout': 30\n}","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Troubleshooting","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Common Issues and Solutions","type":"text"}]},{"type":"paragraph","content":[{"text":"Issue: Out of Memory in Spark Jobs","type":"text","marks":[{"type":"strong"}]}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# Solution: Optimize memory usage\nspark = SparkSession.builder \\\n .config(\"spark.executor.memory\", \"8g\") \\\n .config(\"spark.driver.memory\", \"4g\") \\\n .config(\"spark.sql.shuffle.partitions\", \"200\") \\\n .config(\"spark.default.parallelism\", \"200\") \\\n .getOrCreate()\n\n# Use broadcast joins for small tables\nfrom pyspark.sql.functions import broadcast\nresult = large_df.join(broadcast(small_df), \"key\")","type":"text"}]},{"type":"paragraph","content":[{"text":"Issue: Slow Incremental Loads","type":"text","marks":[{"type":"strong"}]}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# Solution: Use partition pruning and indexing\n# Add indexes on watermark columns\n# Partition target tables by date\n\n# Use partition pruning in Spark\ndf = spark.read.parquet(\"s3://data/table/\") \\\n .where(f\"partition_date >= '{start_date}'\")","type":"text"}]},{"type":"paragraph","content":[{"text":"Issue: Data Quality Failures","type":"text","marks":[{"type":"strong"}]}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# Solution: Implement comprehensive validation\ndef validate_and_quarantine(df, rules):\n \"\"\"Separate valid and invalid records\"\"\"\n \n valid_df = df\n invalid_records = []\n \n for rule_name, rule_func in rules.items():\n mask = rule_func(valid_df)\n invalid = valid_df[~mask].copy()\n invalid['failed_rule'] = rule_name\n invalid_records.append(invalid)\n valid_df = valid_df[mask]\n \n # Save invalid records for review\n if invalid_records:\n pd.concat(invalid_records).to_sql(\n 'data_quality_quarantine',\n conn,\n if_exists='append'\n )\n \n return valid_df","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Best Practices","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Idempotency","type":"text","marks":[{"type":"strong"}]},{"text":": Ensure pipelines can be re-run safely","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Monitoring","type":"text","marks":[{"type":"strong"}]},{"text":": Implement comprehensive logging and alerting","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data Quality","type":"text","marks":[{"type":"strong"}]},{"text":": Validate data at every stage","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Partitioning","type":"text","marks":[{"type":"strong"}]},{"text":": Use appropriate partitioning strategies for performance","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Documentation","type":"text","marks":[{"type":"strong"}]},{"text":": Document data lineage and transformations","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Version Control","type":"text","marks":[{"type":"strong"}]},{"text":": Track schema changes and pipeline versions","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Testing","type":"text","marks":[{"type":"strong"}]},{"text":": Test pipelines with sample data before production","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Security","type":"text","marks":[{"type":"strong"}]},{"text":": Use IAM roles, encryption, and secure credential management","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Interview Preparation","type":"text"}]},{"type":"paragraph","content":[{"text":"Common data engineering interview topics covered:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"SQL optimization and query tuning","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Distributed systems concepts","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data modeling (star schema, snowflake, Data Vault)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ETL vs ELT trade-offs","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"CAP theorem and consistency models","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data quality frameworks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Cloud platform services (S3, Redshift, BigQuery, Databricks)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Orchestration tools (Airflow, Prefect, Dagster)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Streaming architectures (Kafka, Kinesis, Pub/Sub)","type":"text"}]}]}]}]},"metadata":{"date":"2026-11-02","name":"data-engineering-study-material","author":"@skillopedia","source":{"stars":5,"repo_name":"data-skills","origin_url":"https://github.com/reason-machines/data-skills/blob/HEAD/skills/data-engineering-study-material/SKILL.md","repo_owner":"reason-machines","body_sha256":"9a3da4a2b5258f08f85a7f3eb162617f9855eae54b0e01c93f26fa52a31454d0","cluster_key":"e70b2846e8e6d5fc9621210dad571c7bbd190392280cac83df24c8011f22ee17","clean_bundle":{"format":"clean-skill-bundle-v1","source":"reason-machines/data-skills/skills/data-engineering-study-material/SKILL.md","text_bytes":0,"source_commit":"25d6ee79eea935ef4be1ee375a5777b2ad259f0a","skills_sh_rank":8362,"attachment_count":0,"text_attachments":0,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/data-engineering-study-material/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"data-analytics","category_label":"Data"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"data-analytics","triggers":["explain data engineering concepts","show me data engineering study materials","what are data engineering best practices","help me learn data engineering","guide me through data engineering topics","data engineering interview preparation","overview of data engineering tools","data engineering learning resources"],"import_tag":"skills-sh-top20k-2026-11","description":"Comprehensive study guide covering data engineering concepts, tools, and best practices for learning and reference"}},"renderedAt":1793671244656}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.