Join thousands of AI enthusiasts and experts at the Learn AI Community.

Publication

Data Analysis   Data Science   Machine Learning

Big-Data Pipelines with SparkML

Last Updated on January 6, 2023 by Editorial Team

Author(s): Lawrence Alaso Krukrubo

Data Analysis, Data Science, Machineย Learning

Creating Apache Spark ML Pipelines for Big-Dataย Analysis

Photo by Rodion Kutsaev onย Unsplash

Pipelines are a simple way to keep your data preprocessing and modeling code organized. Specifically, a pipeline bundles preprocessing and modeling steps so you can use the whole bundle as if it were a singleย step.

Many Data Scientists hack together models without Pipelinesโ€ฆ (Kaggle)

Kaggle also says Pipelines have some important benefits, suchย as:

  1. Cleaner Code: Accounting for data at each step of preprocessing can get messy. With a Pipeline, you wonโ€™t need to manually keep track of your training and validation data at eachย step.
  2. Fewer Bugs: There are fewer opportunities to misapply a step or forget a preprocessing step.
  3. Easier to Productionize: It can be surprisingly hard to transition a model from a prototype to something deployable at scale, but Pipelines canย help.
  4. More Options for Model Validation: We can easily apply cross-validation and other techniques to our Pipelines.

So, a Pipeline is a convenient process of designing our data preprocessing and Machine Learning flow. There are certain steps that we must do before the actual ML begins. These steps are called data-preprocessing and/or feature engineering.

Some Pipeline stepsย include:

  • Converting categorical values to nominal & numerical values
  • Normalizing the range of values per dimension
  • One-Hot encoding categorical valuesย andโ€ฆ
  • Modelingโ€ฆ where we train our ML algorithm.

The overall idea of Pipelines is that we can fuse our complete data processing flow into one single pipeline, and that single pipeline we can further use downstream.

Some Pipelineย Methods:

Image from IBM scalable-machine-learning-for-big-data-with-apache-spark

Pipeline as a Machine Learning Algorithm has the following methodsโ€ฆ

  • Fit: Fit basically starts theย training
  • Score: Score gives back the predicted value.
  • Evaluate: Evaluates the model performance on the validation data.

One extra advantage of Pipelines is that we can cross-validate, meaning, try out many, many parameters using the very same Pipeline. And this really accelerates the optimization of the algorithm.

So, in summary, pipelines are facilitating our day to day work in machine learning as we can draw from pre-defined data processing steps, we make sure everything is aligned, and we can switch and swap our algorithms asย needed.

While there are plenty of materials covering Pipelines for machine learning, today we shall focus on Pipelines for machine learning on Big-data, using Apacheย SparkML.

1. Intro:

For this exercise, we shall use the HMP dataset. Itโ€™s basically accelerometer recordings from an accelerometer sensor attached to the human body. The data records sensors from humans when performing activities like brush-teeth, comb-hair, eat-soup, and soย on.

So, we shall preprocess this dataset for Machine Learning tasks. First, manually, then we shall build a SparkML Pipeline to preprocess the dataset automatically for us. This Pipeline can be applied to future datasets.

I use Colab for my data exploration. If you need help starting Pyspark in Colab, see thisย link.

So first, we set up Pyspark inย Colabโ€ฆ

2. Data Extraction

Note that the data is in a parquet file format. Parquet uses compression and column store, which maps the data layout to the Apache Spark Tungsten memoryย layout.

So we can see there are different folders in this HMP_dataset. Folders representing different activities such as Brush_teeth, Drink_glass, Getup_bed, Pour_water, Use_telephone.

Looking at the text files in the Brush_teeth folder, for example, we can see the accelerometer data represented in three numeric columns we may call X, Y,ย Z.

!head HMP_Dataset/Brush_teeth/Accelerometer-2011-04-11-13-28-18-brush_teeth-f1.txt
>>
22 49 35
22 49 35
22 52 35
22 52 35
21 52 34

Letโ€™s recursively traverse through these folders in HMP_Dataset and create an Apache spark DataFrame from these text files. Then we just union all DataFrames <df.union(df2)> into one overall DataFrame containing all theย data.

First, Letโ€™s define the schema of the data frameย below.

Now letโ€™s traverse through the data using the OSย library.

Letโ€™s remove non-action folders from the file list. These are typically folders without an underscore in theirย names.

Okay, we have all the folders containing the data in one array. Now we can iterate over thisย array.

So from the Git gist above, first, we define an empty DataFrame and import tqdm for progress-bars. Next, we import the lit library that helps us write String literal columns to an Apache Spark DataFrame.

At this point, all we do is go through each file using the OS library, add the three numeric columns to the X, Y, Z schema we defined earlier, add these to the DataFrame and add two String Literal columns, one for the class of the accelerometer reading and the other for the source file for theย reading.

Letโ€™s see the schema of the DataFrameโ€ฆ

df.printSchema()
>>
root
|-- x: integer (nullable = true)
|-- y: integer (nullable = true)
|-- z: integer (nullable = true)
|-- class: string (nullable = false)
|-- source: string (nullable = false)

Letโ€™s see the first 10 rows of the DataFrame. This takes a little while to run, cos as we know, DataFrames in Apache Spark are alwaysย lazyโ€ฆ

3. Data Transformation

Now we need to transform the data and create an integer representation of the class column as ML algorithms cannot cope with a String. So we will transform the class into a number of integers using the StringIndexer module.

The StringIndexer is an estimator having both fit and transform methods. So we create a StringIndexer object (indexer), pass the โ€˜classโ€™ column as inputCol, and โ€˜classIndexโ€™ as outputCol. Then we fit the DataFrame to the indexer, and transform the DataFrame. This creates a brand new DataFrame (indexed), which we can see above, containing the classIndex additional column.

4. One-Hot Encoding:

With the class index column, we can now do one-hot-encoding inย Pysparkโ€ฆ

Unlike the StringIndexer, OneHotEncoder is a pure transformer, having only the transform method. It uses the same syntax of โ€˜inputColโ€™ and โ€˜outputColโ€™ we saw in StringIndexer. We pass โ€˜classIndexโ€™ and โ€˜categoryVecโ€™ values, respectively. OneHotEncoder also creates a brand new DataFrame (encoded), with a โ€˜category_vecโ€™ column added to the previous DataFrame(indexed).

One more thing to note is that calling encoded.show(10, False) in the Git gist above, ensures that 10 rows are displayed and False ensures that each column element is fully expanded, cos normally SparkML compresses columnย cells.

Finally, OneHotEncoder in SparkML doesn't return several columns containing only zeros and one at a point where a value existed, as we all know… Rather it returns a sparse-vector as seen in the categoryVec column. Thus, for the โ€˜Drink_glassโ€™ class above, SparkML returns a sparse-vector that basically says there are 13 elements, and at position 2, the class value exists(1.0).

5. VectorAssembler:

The next thing we need to do is to transform our numerical columns X, Y, Z into vectors because sparkML can only work on vector objects. So letโ€™s import Vectors and VectorAssembler libraries

The VectorAssembler object gets initialized with the same syntax we used in StringIndexer and OneHotEncoder. We pass list [โ€˜xโ€™, โ€˜yโ€™, โ€˜zโ€™] to inputCols, and we specify outputCol = โ€˜featuresโ€™. This is also a pure transformer like OneHotEncoder. So we transform the DataFrame from the last step (encoded) into a new DataFrame (features_vectorized) with the features columnย added.

The consistency of SparkML syntax is quite impressive, it reduces the learning curve for Big-Data Enthusiastsโ€ฆ

5. Normalizing Theย Dataset:

So, the next step is to normalize the data set. This makes the range of values in the data set for all numerical columns to be between 0 and 1 or -1 and 1. The idea is to have all features data within the same range, so no one overshadows theย other.

We must have gotten used to this by now, StringIndexer, OneHotEncoder, VectorAssembler, and Normalizer all have consistent syntax. Looking at the Normalizer object, it contains the parameter p=1.0. Note that the default norm value for Pyspark Normalizer isย p=2.0.

p=1.0 means the features are normalized based on the Manhattan Distance. Manhattan distance or taxi-cab distance between two points is the sum of the absolute difference between corresponding coordinates of these two points in the feature vectors. So these features are normalized based on thisย metric.

Whileโ€ฆ

p=2.0 means the features are normalized based on the Euclidean Distance. Euclidean distance between two points is exactly the same as computing the magnitude of the vector connecting these twoย points.

Note that these are the same methods used by clustering algorithms like KNN and K-means, forย example.

This is basic to intermediate Linear Algebra, My advice to budding Data Scientists isโ€ฆ Quit chasing each fancy, shiny Algorithm you hear of, rather, spend time to build a solid foundation for Data Science, based on Linear-Algebra, Statistics, Probability and Calculusโ€ฆ

6. Creating The Pipeline:

The Pipeline constructor below takes an array of Pipeline stages we pass to it. Here we pass the 4 stages above in the right sequence, one afterย another.

And that's it! Creating Pipelines in Apache SparkML is as straight as aย rulerโ€ฆ

We define the steps or stages and pass them in a logical sequence to the Pipeline constructor.

Now letโ€™s fit the Pipeline object to our original dataย frameโ€ฆ

data_model = pipeline.fit(df)

Finally, letโ€™s transform our data frame using the Pipelineย Object.

pipelined_data = data_model.transform(df)

Letโ€™s see the first tenย rowsโ€ฆ

So we see that exactly the same DataFrame as created before from the individual stages has been created using the Pipeline function. Now we can fit and transform our data in one go. This is a really handy function.

So at this point, we simply drop the other columns that we donโ€™tย needโ€ฆ

We use a list-comprehension to select the cols we need (categoryVec and features_norm) and simply create a new DataFrame with theseย columns.

So finally, we have our categoryVec column, which is the target variable, and our features_norm column, which is the feature set for the Machine Learning algorithm weโ€™ve been preparing toย train…

Summaryโ€ฆ

Photo by Luca Bravo onย Unsplash

We have seen how to create Apache spark ML Pipelines from our data set. Go out there and use this knowledge to build more robust data and Machine-Learning solutions.

The complete Notebook can be found here onย Github.

Credit goes to The IBM Advanced Data Science Team at Courseraโ€ฆ

Cheers!!

About Me:

Lawrence is a Data Specialist at Tech Layer, passionate about fair and explainable AI and Data Science. I believe that sharing knowledge and experiences is the best way to learn. I hold both the Data Science Professional and Advanced Data Science Professional certifications from IBM and the IBM Data Science Explainability badge, as well as the Artificial Intelligence Nanodegree from Udacity. I have conducted several projects using ML and DL libraries. I love to code up my functions as much as possible. Finally, I never stop learning and exploring, and yes, I have written several highly recommended articles.

Feel free to find meย on:-

Github

Linkedin

Twitter


Big-Data Pipelines with SparkML was originally published in Towards AI on Medium, where people are continuing the conversation by highlighting and responding to this story.

Published via Towards AI

Feedback โ†“