skills.rest
reason-machines
reason-machinesOFFICIAL
Explore all skills by this author →
Website𝕏@reasonmachinesGitHub

spark-data-engineering-pipeline

Builds a PySpark ETL pipeline extracting JSON from S3, validating data, and loading Parquet into PostgreSQL.

5|1|Updated May 16, 2026
Search Tags:#postgresql#data quality#data engineering#parquet#pyspark#aws s3#etl pipeline
One-click install
npx skills add https://github.com/reason-machines/data-skills --skill spark-data-engineering-pipeline-reason-machines
Or copy as Structured Prompt for Agent▼
Please help me install this Agent Skill.
Skill: spark-data-engineering-pipeline
Source: https://github.com/reason-machines/data-skills/tree/main/skills/spark-data-engineering-pipeline
Command: npx skills add https://github.com/reason-machines/data-skills --skill spark-data-engineering-pipeline-reason-machines

SYSTEM DOCUMENTATION & REQUIREMENTS

💡 This Skill requires pyspark, pytest, python-dotenv.

What problem does it solve? Building a production ETL pipeline with PySpark involves wiring together S3 extraction, schema validation, data quality checks, transformations, and database loading, which is error-prone when done from scratch. ## Core Features & Use Cases - End-to-End ETL Flow: Extract JSON data from AWS S3, validate schemas, run data quality checks, transform with window aggregations, write partitioned Parquet, and load into PostgreSQL via JDBC. - Data Quality Framework: Built-in checks for null values, duplicates, value ranges, and categorical validity with a consolidated quality report. - Testing and Troubleshooting: Includes pytest-based unit tests with local Spark sessions plus guidance for S3 connection errors, memory issues, and schema evolution. - Use Case: A data engineer needs to move daily transaction JSON files from an S3 bucket into a PostgreSQL warehouse with cleaning, currency conversion, and per-user aggregations applied along the way. ## Quick Start Ask the AI to set up a PySpark pipeline that reads JSON transaction data from an S3 bucket, validates and cleans it, and loads the result into a PostgreSQL table.

Frequently Asked Questions about spark-data-engineering-pipeline

High-intent search queries and answers about installing and using this skill.

FAQPage Schema
How do I build a PySpark ETL pipeline with S3 and PostgreSQL?▼

Configure a SparkSession with the hadoop-aws and PostgreSQL JDBC packages, read JSON from S3 using the s3a protocol with AWS credentials set in the Hadoop configuration, transform the DataFrame, then write to PostgreSQL using df.write.jdbc with connection properties.

How to validate DataFrame schema in PySpark?▼

Define an expected StructType schema and compare it against the DataFrame's schema by checking field names and data types. Log missing fields, extra fields, and type mismatches, then reject the data if required fields fail validation.

How do I run data quality checks in PySpark?▼

Use DataFrame filters with functions like isNull, dropDuplicates, and isin to count nulls, duplicates, out-of-range values, and invalid categories. Collect the counts into a quality report dictionary for logging and monitoring.

Why does PySpark fail to connect to S3 with s3a?▼

S3 connection failures usually come from missing AWS credentials or a missing hadoop-aws package. Set fs.s3a.access.key and fs.s3a.secret.key in the Hadoop configuration and include org.apache.hadoop:hadoop-aws in spark.jars.packages.

How do I fix out of memory errors in Spark jobs?▼

Repartition large DataFrames to distribute data across more partitions, or coalesce to reduce partition count. You can also increase spark.executor.memory and spark.driver.memory in the SparkSession configuration.

Can PySpark handle schema changes in Parquet files?▼

Yes, use the mergeSchema option when reading Parquet files to handle schema evolution. If a read fails with an AnalysisException due to schema mismatch, retry with spark.read.option("mergeSchema", "true").

Related Skills

Explore More Data & Analytics Skills→
expanso-ioexpanso-io

expanso-s3-to-postgres

Load JSON files from S3 into PostgreSQL with schema mapping and upserts.

Official
Intermediate
aaronachermannaaronachermann

databricks-metric-views

Define and manage YAML-based business metrics in Databricks Unity Catalog.

Community
Advanced
AradotsoAradotso
5

apache-airflow-orchestration

Automate data pipeline scheduling and monitoring with Apache Airflow DAGs.

Official
Advanced
leary-pokenleary-poken
1

databricks-synthetic-data-gen

Generate scalable synthetic data for Databricks workloads using Spark and Faker.

Community
Advanced