Objective

As a herpetologist studying reptiles and amphibians, you have two DataFrames at your disposal: observations (containing sighting logs) and species (containing the reference catalog of animals).

Task

Write a PySpark function that joins the observations and species DataFrames on the species_id column.

After joining, return the top 3 rows ordered by the count of individuals observed in descending order. Save your resulting DataFrame as result_df. Ensure the output matches the exact schema order requested.

File Path

• Observations Dataset: /home/interview/observations.csv
• Species Dataset: /home/interview/species.csv
• Starter script: /home/interview/herpetology.py

Schema

observations.csv

species.csv

Expected Output Schema

Example

Given this sample input:

observations

species

The expected output would be:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("PrepareshSpark").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

observations = spark.read.csv("/home/interview/observations.csv", header=True, inferSchema=True)
species = spark.read.csv("/home/interview/species.csv", header=True, inferSchema=True)

# Step 1: Join the DataFrames on the shared 'species_id' column
joined_df = observations.join(species, on="species_id", how="inner")

# Step 2: Order by count descending and limit to the top 3 rows
top_observations = joined_df.orderBy(F.col("count").desc()).limit(3)

# Step 3: Format the final schema in the exact requested order
result_df = top_observations.select(
    "obs_id", "species_id", "species_name", "location_id", "count"
)

# --- Do not edit below this line ---
result_df.coalesce(1).write.csv("/home/interview/output", header=True, mode="overwrite")
spark.stop()

Explanation

Step 1: Merging the Datasets

joined_df = observations.join(species, on="species_id", how="inner")

To include the species_name in our final output, we need to bring it into the observations DataFrame. We do this by executing an inner join on the shared species_id column.

top_observations = joined_df.orderBy(F.col("count").desc()).limit(3)

The prompt asks for the top 3 rows based on the number of individuals observed. We use .orderBy(F.col("count").desc()) to sort the dataset from the highest count to the lowest. Then, we chain .limit(3) to slice off only the first three rows of the sorted DataFrame, discarding the rest.

Step 3: Output Formatting

result_df = top_observations.select(
    "obs_id", "species_id", "species_name", "location_id", "count"
)

Finally, we use a .select() block to ensure the final output columns are arranged in the precise order specified by the prompt's Expected Output Schema.