Saturday, February 19, 2022

Machine Learning With Spark

 T

Image by Author using Canva.com
  • Machine Learning Basic Concepts
  • Preprocessing and Data Transformation using Spark
  • Spark Clustering with pyspark
  • Classification with pyspark
  • Regression methods with pyspark

What is Apache Spark?

According to Apache Spark and Delta Lake Under the Hood

Image by Author

Setting up Spark 3.0.1 in the Google Colaboratory

As a first step, I configure the google colab runtime with spark installation. For details, readers may read my article Getting Started Spark 3.0.0 in Google Colab om medium.

# Run below commands
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://apache.osuosl.org/spark/spark-3.0.1/spark-3.0.1-bin-hadoop3.2.tgz
!tar xf spark-3.0.1-bin-hadoop3.2.tgz
!pip install -q findspark

Environment Variable

After installing the spark and Java, set the environment variables where Spark and Java are installed.

import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.0.1-bin-hadoop3.2"

Spark Installation test

Let us test the installation of spark in our google colab environment.

import findspark
findspark.init()

from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[*]").getOrCreate()
# Test the spark 
df = spark.createDataFrame([{"hello": "world"} for x in range(1000)])

df.show(3, False)

Machine Learning

Once, we have set up the spark in google colab and made sure it is running with the correct version i.e. 3.0.1 in this case, we can start exploring the machine learning API developed on top of Spark. PySpark is a higher level Python API to use spark with python. For this tutorial, I assume the readers have a basic understanding of Machine Learning and SK-Learn for model building and training. Spark MLlib used the same fit and predict structure as in SK-Learn.

Data Preparation and Transformations in Spark

This section covers the basic steps involved in transformations of input feature data into the format Machine Learning algorithms accept. We will be covering the transformations coming with the SparkML library. To understand or read more about the available spark transformations in 3.0.3, follow the below link.

Extracting, transforming and selecting features

This section covers algorithms for working with features, roughly divided into these groups: Extraction: Extracting…

spark.apache.org

Normalize Numeric Data

MinMaxScaler is one of the favorite classes shipped with most machine learning libraries. It scaled the data between 0 and 1.

from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.linalg import Vectors

Standardize Numeric Data

StandardScaler is another well-known class written with machine learning libraries. It normalizes the data between -1 and 1 and converts the data into bell-shaped data. You can demean the data and scale to some variance.

from pyspark.ml.feature import  StandardScaler
from pyspark.ml.linalg import Vectors

Bucketize Numeric Data

The real data sets come with various ranges and sometimes it is advisable to transform the data into well-defined buckets before plugging into machine learning algorithms.

from pyspark.ml.feature import  Bucketizer
from pyspark.ml.linalg import Vectors

Tokenize text Data

Natural Language Processing is one of the main applications of Machine learning. One of the first steps for NLP is tokenizing the text into words or token. We can utilize the Tokenizer class with SparkML to perform this task.

from pyspark.ml.feature import  Tokenizer

TF-IDF

Term frequency-inverse document frequency (TF-IDF) is a feature vectorization method widely used in text mining to reflect the importance of a term to a document in the corpus. Using the above-tokenized data, Let us apply the TF-IDF

from pyspark.ml.feature import HashingTF, IDF

Clustering Using PySpark

Clustering is a machine learning technique where the data is grouped into a reasonable number of classes using the input features. In this section, we study the basic application of clustering techniques using the spark ML framework.

from pyspark.ml.linalg import Vectors
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.clustering import KMeans, BisectingKMeans
import glob
# Read the data.
clustering_file_name ='clustering_dataset.csv'
import pandas as pd
# df = pd.read_csv(clustering_file_name)
cluster_df = spark.read.csv(clustering_file_name, header=True,inferSchema=True)
# Coverting the input data into features column
vectorAssembler = VectorAssembler(inputCols = ['col1', 'col2', 'col3'], outputCol = "features")
vcluster_df = vectorAssembler.transform(cluster_df)
# Applying the k-means algorithm
kmeans = KMeans().setK(3)
kmeans = kmeans.setSeed(1)
kmodel = kmeans.fit(vcluster_df)
centers = kmodel.clusterCenters()
print("The location of centers: {}".format(centers))
# Applying Hierarchical Clustering
bkmeans = BisectingKMeans().setK(3)
bkmeans = bkmeans.setSeed(1)

Clustering

This page describes clustering algorithms in MLlib. The guide for clustering in the RDD-based API also has relevant…

spark.apache.org

Classification Using PySpark

Classification is one of the widely used Machine algorithms and almost every data engineer and data scientist must know about these algorithms. Once the data is loaded and prepared, I will demonstrate three classification algorithms.

  1. Multi-Layer Perceptron Classification
  2. Decision Trees Classification
# Downloading the clustering data
!wget -q "https://raw.githubusercontent.com/amjadraza/blogs-data/master/spark_ml/iris.csv"
png
spark.createDataFrame(df, columns)

Preprocessing the Iris Data

In this section, we will be using the IRIS data to understand the classification. To perform ML models, we apply the preprocessing step on our input data.

from pyspark.sql.functions import *
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.feature import  StringIndexer

Naive Bayes Classification

Once the data is prepared, we are ready to apply the first classification algorithm.

from pyspark.ml.classification import NaiveBayes
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="accuracy")
nbaccuracy = evaluator.evaluate(predictions_df)
nbaccuracy

Multilayer Perceptron Classification

The second classifier we will be investigating is a Multi-layer perceptron. In this tutorial, I am not going into details of the optimal MLP network for this problem however in practice, you research the optimal network suitable to the problem in hand.

from pyspark.ml.classification import MultilayerPerceptronClassifier

Decision Trees Classification

Another common classifier in the ML family is the Decision Tree Classifier, in this section, we explore this classifier.

from pyspark.ml.classification import DecisionTreeClassifier

Classification and regression

\newcommand{\R}{\mathbb{R}} \newcommand{\E}{\mathbb{E}} \newcommand{\x}{\mathbf{x}} \newcommand{\y}{\mathbf{y}}…

spark.apache.org

Regression using PySpark

In this section, we explore the Machine learning models for regression problems using pyspark. Regression models are helpful in predicting future values using past data.

from pyspark.ml.regression import LinearRegression
from pyspark.ml.feature import VectorAssembler

Linear Regression

We start with the simplest regression technique i.e. Linear Regression.

# Define and fit Linear Regression
lr = LinearRegression(featuresCol="features", labelCol="PE")
lr_model = lr.fit(vpp_df)

Decision Tree Regression

In this section, we explore the Decision Tree Regression commonly used in Machine learning.

from pyspark.ml.regression import DecisionTreeRegressor
from pyspark.ml.evaluation import RegressionEvaluator

Gradient Boosting Decision Tree Regression

Gradient Boosting is another common choice among ML professionals. Let us try the GBM in this section.

from pyspark.ml.regression import GBTRegressor

Classification and regression

\newcommand{\R}{\mathbb{R}} \newcommand{\E}{\mathbb{E}} \newcommand{\x}{\mathbf{x}} \newcommand{\y}{\mathbf{y}}…

spark.apache.org

A working Google Colab

Conclusions

In this tutorial, I have tried to give the readers an opportunity to learn and implement basic Machine Learning algorithms using PySpark. Spark not only provide the benefit of distributed processing but also can handle a large amount of data to be processing. To summarise, we have covered below topics/algorithms

  • Overview of Data Transformations using PySpark
  • Clustering algorithms using PySpark
  • Classification problems using PySpark
  • Regression Problems using PySpark

References Readings/Links

  1. https://spark.apache.org/docs/latest/ml-features.html
  2. https://spark.apache.org/docs/3.0.1/ml-classification-regression.html#regression
  3. https://spark.apache.org/docs/3.0.1/ml-clustering.html
  4. https://spark.apache.org/docs/3.0.1/ml-classification-regression.html#classification



at No comments:

Saturday, February 5, 2022

Kafka with CDC and Delta Lake’s CDF

 This is a second part of the Data Lakehouse and data pipelines implementation in the Delta Lake. Source code GitHub repositories are at the end of this article. For a high level pipeline architecture, please take a look at the first part.

Road to Lakehouse — Part 1: Delta Lake data pipeline overview

Raw Ingestion

I divide the Kafka data into 2 categories: event data which comes from the backend application and cdc data which is generated by Debezium. Below is a main PySpark job.

Ingest data from Kafka

To ingest the data from Kafka, we just need to specify credentials and topic name.

Process the Kafka data

We need to get the schema of the topic first.

In case of CDC, below is an example of MongoDB payload schema.

Once we have the schema, it is easy to get the data in plain texts.

Load the processed data into a raw area

Eventually let write the stream to a delta table in the raw zone. Spark stores Kafka offsets in checkpoint locations for failure recoveries.

Refined zone

While there are several ways to transform data inside the Delta Lake between different layers such as using dbt or Delta Live Tables, we can leverage the built-in property in delta tables which is the Change Data Feed (CDF) without any additional cost to merge changes to the next area in near real-time.

Below is a main PySpark job to extract changes from the raw table then process, load them into the refined table.

Read change feed from raw tables

Process the CDF

We can do some transformations like flattening a nested json or exploding an array before loading the data into the refined layer.

Conclusion

Above streaming data pipeline is a good starting point to build further business level tables. In your real project you might need to do more complicated operations like upserting or joining multiple tables, but the general idea of using Spark structured streaming and CDF is still valid. There are many more things to do with the Delta Lake like using SQLAlchemy ORM to query data, visualizing data with Streamlit and building ML workflows with Databricks Feature Store, AutoML, MLflow, .etc, hopefully we can discuss more about them in the next article.

Source code

GitHub - tam159/delta-lake-library: Delta lake common libraries to ingest, process and load data…

Data Python artifact wheel file is installed in the Spark cluster and can be used in all cluster notebooks. The package…

github.com

GitHub - tam159/delta-lake-pipeline: Delta lake pipelines in Databricks lakehouse with Spark…

You can't perform that action at this time. You signed in with another tab or window. You signed out in another tab or…

github.com

at No comments:

Road to Lakehouse - Part 1: Delta Lake data pipeline overview

 

Delta Lake Overview
Delta Lake Overview
Tam Nguyen

Tam Nguyen

Data Analytics | Big Data Engineer | Lakehouse | MLOps
2 articles 

This is a first part of the Data Lakehouse and data pipelines implementation in the Delta Lake. Source code GitHub repositories are at the end of this article. For the code explanation, please take a look at the second part.

Road to Lakehouse - Part 2: Ingest and process data from Kafka with CDC and Delta Lake's CDF

What and why Data Lakehouse?

Lakehouse is a buzzword in the data field nowadays. According to AWS, the lake house architecture is about integrating a data lake, a data warehouse, and purpose-built stores to enable unified governance and easy data movement. Regarding to Databricks, the Lakehouse is an open architecture that combines the best elements of data lakes and data warehouses.

And for me, the Lakehouse is simply the data lake with both columnar files and delta files (transaction logs) in row based formats.

Delta Lake, Iceberg and Hudi are 3 popular data lake table formats that support ACID, schema evolution, upsert, time travel, incremental consumption, etc.

In this post, we will focus on the Delta Lake with support from Databricks.

Data lakehouse features

In comparison with the pure data lake, the Lakehouse provides:

  • ACID transactions to ensure data integrity and consistency as multiple parties concurrently read or write data.
  • Schema enforcement and evolution that we can optionally raise an error when data source schema changes or automatically merge it with the new one.
  • Time travel or data versioning allows us to access and revert to earlier versions of data, rollbacks or reproduce experiments.
  • Full DML supports UPDATE, DELETE and especially MERGE INTO is really useful in case of using with CDC and Delta Lake Change Data Feed (CDF).
  • End-to-end streaming eliminates the need for separating systems dedicated to serve real-time data applications because Delta Lake table is both a batch as well as a streaming source and sink

The Lakehouse also has more features compared with the traditional Data Warehouse:

  • Openness: The storage formats are open and standardized such as Apache Parquet which enables Delta tables to be queryable by different tools with or without Spark like Presto, Trino and different languages including Scala, Java, Rust, Python, Ruby, .etc
  • Support for diverse data types ranging from unstructured to structured data, so we can not only store structured, semi-structured data and text in the Lakehouse but also binary files like images, video, audio, .etc
  • Support for diverse workloads including data science, machine learning, and SQL analytics. It means that we do not need to move the data between data lake and data warehouse for training data, one Lakehouse destination is nearly enough for almost all purposes.

There are some highlights of Delta Lake:

  • Delta Sharing secures the way to share data with other organizations regardless of which computing platforms they use.
  • Data optimization including compaction (coalescing small files into larger ones) and ordering.
  • Vacuum: remove data files that are no longer in the latest state of the transaction log for the table and are older than a retention threshold.
  • Table caching in the Spark nodes’ local storage.


Data pipeline overview

For a demonstration, we will implement a below data pipeline in the Delta lake.

Delta Lake Data Pipeline

  1. Regarding to data sources, there can be multiple sources such as:

  • OLTP systems logs can be captured and sent to Apache Kafka (1a)
  • Third party tools can also be integrated with the Kafka through various connectors (1b)
  • And back-end team can produce event data directly to the Kafka (1c)

2. Once we have the data in the Kafka, we will use Spark Structured Streaming to consume the data and load it into Delta tables in a raw area. Thanks to checkpointing which stores Kafka offsets in this case, we can recover from failures with exactly-once fault-tolerant. We need to enable the Delta Lake CDF feature in these raw tables in order to serve further layers.

3. We will use the Spark Structured Streaming again to ingest changes from the raw tables. Then we would do some transformations like flattening and exploding nested data, .etc and load the cleansed data to a next area which is a refined zone. Remember to add the CDF in the refined tables properties.

4. Now we are ready to build data mart tables for business level by aggregating or joining tables from the refined area. This step is still in near real-time process because one more time we read the changes from previous layer tables by Spark Structured Streaming.

5. All the metadata is stored in the Databricks Data Catalog and all above tables can be queried in Databricks SQL Analytics, where we can create SQL endpoints and use a SQL editor. Beside the default catalog, we can use an open source Amundsen for the metadata discovery.

6. Eventually we can build some data visualizations from Databricks SQL Analytics Dashboards (formerly Redash) or use BI tools like Power BI, Streamlit, .etc.


Conclusion

The Lakehouse provides a comprehensive architecture that benefits from both Data Lake and Data Warehouse. Just with the straight data pipeline, we can serve for analytical and operational purposes. In the next part, we will ingest streaming data from MongoDB with Debezium and back-end events then load it into the first zone of the Lakehouse, after that we will process changes from the raw area and store cleansed and flattened data in a refined zone.

at No comments:
Subscribe to: Posts (Atom)

Deduplicating Data on the Databricks Lakehouse: Making joins, BI, and AI queries “safe by default.”

  Imagine this: your manager asks the AI analytics tool: "What were our top-selling products last quarter?" The AI generates perfe...