ML Pipelines in Spark in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of a simple Spark ML Pipeline

What is the output of the following Spark ML Pipeline code snippet when applied to the given DataFrame?

Apache Spark

from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, HashingTF
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
data = spark.createDataFrame([
    (0, "I love Spark"),
    (1, "Spark is great"),
    (2, "I enjoy learning ML")
], ["id", "sentence"])

tokenizer = Tokenizer(inputCol="sentence", outputCol="words")
hashingTF = HashingTF(inputCol="words", outputCol="features", numFeatures=10)
pipeline = Pipeline(stages=[tokenizer, hashingTF])
model = pipeline.fit(data)
result = model.transform(data).select("id", "features").collect()

A [Row(id=0, features=SparseVector(10, {0: 1.0, 2: 1.0, 4: 1.0})), Row(id=1, features=SparseVector(10, {0: 1.0, 2: 1.0, 6: 1.0})), Row(id=2, features=SparseVector(10, {0: 1.0, 1: 1.0, 3: 1.0}))]

B [Row(id=0, features=SparseVector(10, {0: 1.0, 3: 1.0, 5: 1.0})), Row(id=1, features=SparseVector(10, {0: 1.0, 4: 1.0, 6: 1.0})), Row(id=2, features=SparseVector(10, {1: 1.0, 2: 1.0, 7: 1.0}))]

C [Row(id=0, features=SparseVector(10, {1: 1.0, 3: 1.0, 6: 1.0})), Row(id=1, features=SparseVector(10, {1: 1.0, 3: 1.0, 7: 1.0})), Row(id=2, features=SparseVector(10, {1: 1.0, 2: 1.0, 5: 1.0}))]

D [Row(id=0, features=SparseVector(10, {1: 1.0, 4: 1.0, 7: 1.0})), Row(id=1, features=SparseVector(10, {1: 1.0, 5: 1.0, 8: 1.0})), Row(id=2, features=SparseVector(10, {2: 1.0, 3: 1.0, 6: 1.0}))]

💡 Hint

Focus on how Tokenizer splits sentences and how HashingTF maps words to feature indices.

Tokenizer splits each sentence into words. HashingTF converts these words into a sparse vector of fixed length 10 using a hash function. The indices and values in the SparseVector correspond to hashed word features.

❓ data_output

intermediate

Number of stages in a Spark ML Pipeline

Given the following pipeline construction code, how many stages does the pipeline contain?

Apache Spark

from pyspark.ml.feature import StringIndexer, VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml import Pipeline

indexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
assembler = VectorAssembler(inputCols=["feature1", "feature2", "categoryIndex"], outputCol="features")
lr = LogisticRegression(featuresCol="features", labelCol="label")

pipeline = Pipeline(stages=[indexer, assembler, lr])
num_stages = len(pipeline.getStages())

A 2

B 3

C 4

D 1

💡 Hint

Count each transformer and estimator added to the stages list.

The pipeline has three stages: StringIndexer, VectorAssembler, and LogisticRegression. Each is a stage in the pipeline.

🔧 Debug

advanced

Identify the error in this Spark ML Pipeline code

What error will this Spark ML Pipeline code raise when executed?

Apache Spark

from pyspark.ml.feature import Tokenizer, HashingTF
from pyspark.ml import Pipeline

spark = SparkSession.builder.getOrCreate()
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol="words", outputCol="features")
pipeline = Pipeline(stages=[tokenizer, hashingTF])

# DataFrame missing 'text' column
data = spark.createDataFrame([(0, "hello world")], ["id", "sentence"])
model = pipeline.fit(data)

A No error, pipeline fits successfully

B TypeError: stages must be a list of transformers and estimators

C AnalysisException: cannot resolve '`text`' given input columns: [id, sentence]

D ValueError: inputCol and outputCol must be different

💡 Hint

Check if the input column names in the pipeline match the DataFrame columns.

The Tokenizer expects an input column named 'text', but the DataFrame has 'sentence'. This mismatch causes an AnalysisException.

🚀 Application

advanced

Choosing the correct pipeline stage order

You want to build a Spark ML Pipeline to classify text data. Which pipeline stage order is correct?

A Tokenizer -> HashingTF -> LogisticRegression

B LogisticRegression -> Tokenizer -> HashingTF

C LogisticRegression -> HashingTF -> Tokenizer

D HashingTF -> Tokenizer -> LogisticRegression

💡 Hint

Think about the data flow: raw text to tokens, tokens to features, features to model.

The correct order is to first tokenize the text, then convert tokens to features, and finally apply the classifier.

🧠 Conceptual

expert

Effect of caching in Spark ML Pipelines

What is the main benefit of caching intermediate DataFrames in a Spark ML Pipeline during iterative model tuning?

A It reduces the time to recompute transformations by storing intermediate results in memory.

B It automatically improves model accuracy by using cached data.

C It prevents the pipeline from executing stages in parallel.

D It increases disk usage by saving all intermediate data permanently.

💡 Hint

Consider how caching affects repeated computations in Spark.

Caching stores intermediate DataFrames in memory, so repeated computations during tuning reuse these results, speeding up processing.

Practice

1. What is the main purpose of using ML Pipelines in Apache Spark?

easy

A. To visualize data in charts and graphs

B. To chain multiple data processing and machine learning steps into one workflow

C. To store data permanently in a database

D. To write SQL queries for data extraction

Solution

1. Step 1: Understand ML Pipelines concept

   ML Pipelines are designed to combine multiple steps like data cleaning, feature extraction, and model training into a single workflow.

2. Step 2: Identify the main purpose

   This chaining helps automate the process and apply the same steps easily to new data.

3. Final Answer:

   To chain multiple data processing and machine learning steps into one workflow -> Option B

4. Quick Check:

   ML Pipelines = chaining steps [OK]

Hint: Remember: Pipelines automate and chain ML steps [OK]

Common Mistakes:

• Confusing pipelines with data visualization tools
• Thinking pipelines store data permanently
• Assuming pipelines are only for SQL queries

2. Which of the following is the correct way to create a Pipeline in Spark ML?

easy

A. pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])

B. pipeline = Pipeline(steps=[tokenizer, hashingTF, lr])

C. pipeline = Pipeline(chain=[tokenizer, hashingTF, lr])

D. pipeline = Pipeline(process=[tokenizer, hashingTF, lr])

Solution

1. Step 1: Recall Spark ML Pipeline syntax

   In Spark ML, the Pipeline constructor expects a parameter named stages which is a list of transformers and estimators.

2. Step 2: Match the correct parameter name

   Only pipeline = Pipeline(stages=[tokenizer, hashingTF, lr]) uses stages correctly; others use incorrect parameter names.

3. Final Answer:

   pipeline = Pipeline(stages=[tokenizer, hashingTF, lr]) -> Option A

4. Quick Check:

   Use 'stages' to list pipeline steps [OK]

Hint: Use 'stages' parameter to add steps in Pipeline [OK]

Common Mistakes:

• Using 'steps' instead of 'stages'
• Using 'chain' or 'process' which are invalid
• Forgetting to pass a list of stages

3. Given the following Spark ML Pipeline code, what will be the output type of model.transform(testData)?

from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, HashingTF
from pyspark.ml.classification import LogisticRegression

# Define stages
tokenizer = Tokenizer(inputCol='text', outputCol='words')
hashingTF = HashingTF(inputCol='words', outputCol='features')
lr = LogisticRegression(maxIter=10, regParam=0.01)

pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])

# Fit the pipeline
model = pipeline.fit(trainingData)

# Transform test data
result = model.transform(testData)
print(type(result))

medium

A.

B.

C.

D.

Solution

1. Step 1: Understand what model.transform returns

   In Spark ML, calling transform() on a PipelineModel returns a DataFrame with new columns added by the pipeline stages.

2. Step 2: Identify the output type

   The output is a Spark DataFrame, not a list, dict, or PipelineModel.

3. Final Answer:

   <class 'pyspark.sql.dataframe.DataFrame'> -> Option D

4. Quick Check:

   model.transform() returns DataFrame [OK]

Hint: transform() always outputs a DataFrame in Spark ML [OK]

Common Mistakes:

• Thinking transform returns the model itself
• Expecting a Python list or dict instead of DataFrame
• Confusing fit() output with transform() output

4. Identify the error in the following Spark ML Pipeline code snippet:

from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, HashingTF

tokenizer = Tokenizer(inputCol='text', outputCol='words')
hashingTF = HashingTF(inputCol='words', outputCol='features')

pipeline = Pipeline(stages=[tokenizer, hashingTF])
model = pipeline.fit(trainingData)
result = model.transform(testData)
print(result.select('prediction').show())

medium

A. Missing an estimator stage like LogisticRegression before fitting

B. Tokenizer inputCol should be 'words' instead of 'text'

C. Pipeline stages should be passed as a tuple, not a list

D. transform() cannot be called on the pipeline model

Solution

1. Step 1: Check pipeline stages

   The pipeline only has transformers (Tokenizer and HashingTF) but no estimator (like LogisticRegression) to train a model.

2. Step 2: Understand fit() behavior

   Fitting a pipeline without an estimator returns a PipelineModel but no trained model for prediction; this is valid but usually incomplete for ML tasks.

3. Step 3: Identify the error

   For supervised learning, missing an estimator means no model is trained, so the pipeline is incomplete.

4. Final Answer:

   Missing an estimator stage like LogisticRegression before fitting -> Option A

5. Quick Check:

   Estimator needed for training model [OK]

Hint: Always include an estimator for training in pipeline stages [OK]

Common Mistakes:

• Confusing inputCol names
• Thinking stages must be tuples
• Believing transform() can't be called on PipelineModel

5. You want to create a Spark ML Pipeline that preprocesses text data, trains a logistic regression model, and then applies the same pipeline to new streaming data. Which approach is best to ensure the pipeline works correctly on both batch and streaming data?

hard

A. Use only Tokenizer and HashingTF in the pipeline; train LogisticRegression on streaming data directly

B. Train LogisticRegression separately on batch data; apply Tokenizer and HashingTF manually on streaming data

C. Build a Pipeline with Tokenizer, HashingTF, LogisticRegression; fit on batch data; save and load the PipelineModel for streaming transform

D. Fit the pipeline on streaming data first, then apply it to batch data

Solution

1. Step 1: Understand pipeline reuse for batch and streaming

   To apply the same transformations and model to streaming data, you must save the trained PipelineModel from batch training.

2. Step 2: Best practice for streaming

   Fit the pipeline on batch data, save the PipelineModel, then load it in streaming jobs to transform streaming data consistently.

3. Step 3: Evaluate options

   Build a Pipeline with Tokenizer, HashingTF, LogisticRegression; fit on batch data; save and load the PipelineModel for streaming transform follows this best practice. Other options either separate steps manually or train on streaming data first, which is less reliable.

4. Final Answer:

   Build a Pipeline with Tokenizer, HashingTF, LogisticRegression; fit on batch data; save and load the PipelineModel for streaming transform -> Option C

5. Quick Check:

   Save and reuse PipelineModel for streaming [OK]

Hint: Train pipeline on batch, save model, reuse for streaming [OK]

Common Mistakes:

• Training model separately from preprocessing
• Trying to train on streaming data first
• Not saving/loading PipelineModel for reuse