Integration testing pipelines in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of Spark DataFrame after pipeline transformation

Given the following Spark pipeline code, what is the output DataFrame content after applying the transformations?

Apache Spark

from pyspark.sql import SparkSession
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler

spark = SparkSession.builder.master('local').appName('Test').getOrCreate()
data = [(0, 'cat'), (1, 'dog'), (2, 'cat'), (3, 'bird')]
schema = ['id', 'animal']
df = spark.createDataFrame(data, schema)

indexer = StringIndexer(inputCol='animal', outputCol='animalIndex')
assembler = VectorAssembler(inputCols=['animalIndex'], outputCol='features')
pipeline = Pipeline(stages=[indexer, assembler])
model = pipeline.fit(df)
result = model.transform(df).select('id', 'animal', 'animalIndex', 'features')
result.show()

A [Row(id=0, animal='cat', animalIndex=0.0, features=DenseVector([0.0])), Row(id=1, animal='dog', animalIndex=1.0, features=DenseVector([1.0])), Row(id=2, animal='cat', animalIndex=0.0, features=DenseVector([0.0])), Row(id=3, animal='bird', animalIndex=2.0, features=DenseVector([2.0]))]

B [Row(id=0, animal='cat', animalIndex=1.0, features=DenseVector([0.0])), Row(id=1, animal='dog', animalIndex=2.0, features=DenseVector([1.0])), Row(id=2, animal='cat', animalIndex=1.0, features=DenseVector([0.0])), Row(id=3, animal='bird', animalIndex=0.0, features=DenseVector([2.0]))]

C [Row(id=0, animal='cat', animalIndex=2.0, features=DenseVector([2.0])), Row(id=1, animal='dog', animalIndex=0.0, features=DenseVector([0.0])), Row(id=2, animal='cat', animalIndex=2.0, features=DenseVector([2.0])), Row(id=3, animal='bird', animalIndex=1.0, features=DenseVector([1.0]))]

D [Row(id=0, animal='cat', animalIndex=1.0, features=DenseVector([1.0])), Row(id=1, animal='dog', animalIndex=2.0, features=DenseVector([2.0])), Row(id=2, animal='cat', animalIndex=1.0, features=DenseVector([1.0])), Row(id=3, animal='bird', animalIndex=0.0, features=DenseVector([0.0]))]

💡 Hint

Remember StringIndexer assigns indices based on label frequency descending order.

StringIndexer assigns indices starting from 0 for the most frequent label. 'cat' appears twice, so it gets index 0 because it is the most frequent. 'dog' appears once and 'bird' appears once, so they get indices 1 and 2 respectively based on label ordering. VectorAssembler creates a vector from the animalIndex column.

❓ data_output

intermediate

Number of rows after filtering in Spark pipeline

Consider a Spark DataFrame with 1000 rows. A pipeline stage filters rows where the column 'age' is greater than 30. If 400 rows have age > 30, how many rows remain after the pipeline transformation?

Apache Spark

from pyspark.sql import SparkSession
from pyspark.ml import Pipeline
from pyspark.sql.functions import col

spark = SparkSession.builder.master('local').appName('Test').getOrCreate()
data = [(i, 20 + (i % 50)) for i in range(1000)]
schema = ['id', 'age']
df = spark.createDataFrame(data, schema)

class AgeFilter:
    def __init__(self, age_limit):
        self.age_limit = age_limit
    def transform(self, df):
        return df.filter(col('age') > self.age_limit)
    def fit(self, df):
        return self

filter_stage = AgeFilter(30)
pipeline = Pipeline(stages=[filter_stage])
model = pipeline.fit(df)
result = model.transform(df)
count = result.count()

A 1000

B 400

C 600

D 0

💡 Hint

Count how many rows satisfy the condition age > 30.

Since ages cycle from 20 to 69, exactly 400 rows have age > 30, so after filtering, 400 rows remain.

🔧 Debug

advanced

Identify the error in Spark pipeline code

What error does the following Spark pipeline code raise when executed?

Apache Spark

from pyspark.sql import SparkSession
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer

spark = SparkSession.builder.master('local').appName('Test').getOrCreate()
data = [(0, 'apple'), (1, 'banana')]
schema = ['id', 'fruit']
df = spark.createDataFrame(data, schema)

indexer = StringIndexer(inputCol='fruit', outputCol='fruitIndex')
pipeline = Pipeline(stages=[indexer])
model = pipeline.fit(df)
result = model.transform(df).select('id', 'fruitIndex')
result.show()

A No error, outputs DataFrame with 'id' and 'fruitIndex' columns

B TypeError: 'StringIndexer' object is not callable

C AnalysisException: cannot resolve 'fruitIndex' given input columns: [id, fruit]

D AttributeError: 'Pipeline' object has no attribute 'fit'

💡 Hint

Check if the pipeline stages are correctly defined and used.

The code correctly fits the pipeline and transforms the DataFrame. The output DataFrame contains 'id' and 'fruitIndex' columns without error.

🚀 Application

advanced

Best approach to test Spark pipeline integration

Which approach is best to test the integration of multiple Spark pipeline stages in a data science project?

A Only test the pipeline on production data to ensure real-world correctness

B Test each pipeline stage separately with unit tests and skip full pipeline testing

C Run the full pipeline on a small sample dataset and verify the final output matches expected results

D Manually inspect intermediate DataFrame outputs without automated tests

💡 Hint

Integration testing means testing combined parts together.

Integration testing verifies that all pipeline stages work together correctly by running the full pipeline on sample data and checking outputs.

🧠 Conceptual

expert

Effect of caching in Spark pipeline integration tests

In Spark pipeline integration tests, what is the main benefit of caching intermediate DataFrames during the pipeline execution?

A It speeds up repeated access to intermediate results, reducing test runtime

B It automatically fixes data quality issues in intermediate DataFrames

C It prevents the pipeline from running any transformations

D It increases memory usage without any performance benefit

💡 Hint

Think about how caching affects performance in Spark.

Caching stores intermediate results in memory, so repeated accesses during tests are faster, reducing overall test time.