Implement Election Report and Banking Pipeline

You are given two independent data engineering coding tasks.

Task 1: SQL Election Exit Poll by State Report

An election analytics team needs a report showing, for each candidate, the top three vote totals by state and the states where those totals occurred.

Tables

candidates

results

Each row in results represents one vote for a candidate in a state.

Output

Return one row per candidate with the following columns:

Each place column should contain one or more values formatted as:

State Name (votes)

If multiple states have the same vote count for the same place, sort those states alphabetically and join them with , .

For example:

Alabama (23), New York (23)

The top three places are based on distinct vote totals, not on individual states. For example, if a candidate has state vote counts 3, 3, 2, 1, then the first place contains both states with 3, the second place contains the state with 2, and the third place contains the state with 1.

Sample Data

candidates

results

Expected Output

Write a SQL query that produces this report.

Task 2: PySpark Banking Data Mining

Implement a PySpark job that reads account and transaction CSV files, filters valid transactions, and computes account-level transaction statistics.

Input Files

accounts.csv

One row per account.

transactions.csv

One row per transfer transaction.

Rules

A transaction is valid if all of the following are true:

1. fromAccountNumber exists in accounts.csv.
2. toAccountNumber exists in accounts.csv.
3. transferAmount is less than or equal to the balance of the source account.

Methods to Implement

You are given a partially completed class main.job.pipeline.PySparkJob. Implement the following methods:

class PySparkJob:
 def init_spark_session(self) -> SparkSession:
 """Create and return a SparkSession with master 'local' and app name 'Banking Data Mining'."""
 pass

 def extract_valid_transactions(self, accounts: DataFrame, transactions: DataFrame) -> DataFrame:
 """Return only valid transactions according to the validation rules above."""
 pass

 def distinct_transactions(self, transactions: DataFrame) -> int:
 """Return the number of distinct source accounts, based on fromAccountNumber."""
 pass

 def transactions_per_account(self, transactions: DataFrame) -> dict:
 """Return the top 10 fromAccountNumber values by transaction count as a dictionary."""
 pass

The helper method read_csv(file_path) reads CSV files with headers. Unit tests are provided and must pass.