ScaDaMaLe Course site and book

Introduction

• Course Name: Scalable Data Science and Distributed Machine Learning
• Course Acronym: ScaDaMaLe or sds-3.x.

The course is given in several modules.

Expected Reference Readings

Note that you need to be logged into your library with access to these publishers:

• https://learning.oreilly.com/library/view/high-performance-spark/9781491943199/
• https://learning.oreilly.com/library/view/spark-the-definitive/9781491912201/
• https://learning.oreilly.com/library/view/learning-spark-2nd/9781492050032/
• Introduction to Algorithms, Third Edition, Thomas H. Cormen, Charles E. Leiserson, Ronald L. Rivest, and Clifford Stein from
  • https://ebookcentral.proquest.com/lib/uu/reader.action?docID=3339142
• Reading Materials Provided

Course Sponsors

The course builds on contents developed since 2016 with support from New Zealand's Data Industry. The 2017-2019 versions were academically sponsored by Uppsala University's Inter-Faculty Course grant, Department of Mathematics and The Centre for Interdisciplinary Mathematics and industrially sponsored by databricks, AWS and Swedish data industry via Combient AB, SEB and Combient Mix AB. The 2021-2023 versions were/are academically sponsored by WASP Graduate School and Centre for Interdisciplinary Mathematics, and industrially sponsored by databricks and AWS via databricks University Alliance and Combient Mix AB via industrial mentorships and internships.

Course Instructor

I, Raazesh Sainudiin or Raaz, will be an instructor for the course.

I have

• more than 16 years of academic research experience in applied mathematics and statistics and
• over 8 years of experience in the data industry.

I currently (2022) have an effective joint appointment as:

• Associate Professor of Mathematics with specialisation in Data Science at Department of Mathematics, Uppsala University, Uppsala, Sweden and
• Researching,Developing and Enabling Chair in Mathematical Data Engineering Sciences at Combient Mix AB, Stockholm, Sweden

Quick links on Raaz's background:

• https://www.linkedin.com/in/raazesh-sainudiin-45955845/
• Raaz's academic CV
• Raaz's publications list

What is the Data Science Process

The Data Science Process in one picture

Data Science Process under the Algorithms-Machines-Peoples-Planet Framework

Note that the Data Product that is typically a desired outcome of the Data Science Process can be anything that has commercial value (to help make a livig, colloquially speaking), including a software product, hardware product, personalized medicine for a specific individual, or pasture-raised chicken based on intensive data collection from field experiments in regenerative agriculture, among others.

It is extremely important to be aware of the underlying actual costs and benefits in any Data Science Process under the Algorithms-Machines-Peoples-Planet Framework. We will see this in the sequel at a high level.

What is scalable data science and distributed machine learning?

Scalability merely refers to the ability of the data science process to scale to massive datasets (popularly known as big data).

For this we need distributed fault-tolerant computing typically over large clusters of commodity computers -- the core infrastructure in a public cloud today.

Distributed Machine Learning allows the models in the data science process to be scalably trained and extract value from big data.

What is Data Science?

It is increasingly accepted that Data Science

is an inter-disciplinary field that uses scientific methods, processes, algorithms and systems to extract knowledge and insights from many structural and unstructured data. Data science is related to data mining, machine learning and big data.

Data science is a "concept to unify statistics, data analysis and their related methods" in order to "understand and analyze actual phenomena" with data. It uses techniques and theories drawn from many fields within the context of mathematics, statistics, computer science, domain knowledge and information science. Turing award winner Jim Gray imagined data science as a "fourth paradigm" of science (empirical, theoretical, computational and now data-driven) and asserted that "everything about science is changing because of the impact of information technology" and the data deluge.

Now, let us look at two industrially-informed academic papers that influence the above quote on what is Data Science, but with a view towards the contents and syllabus of this course.

Source: Vasant Dhar, Data Science and Prediction, Communications of the ACM, Vol. 56 (1). p. 64, DOI:10.1145/2500499

key insights in the above paper

• Data Science is the study of the generalizabile extraction of knowledge from data.
• A common epistemic requirement in assessing whether new knowledge is actionable for decision making is its predictive power, not just its ability to explain the past.
• A data scientist requires an integrated skill set spanning
  • mathematics,
  • machine learning,
  • artificial intelligence,
  • statistics,
  • databases, and
  • optimization,
  • along with a deep understanding of the craft of problem formulation to engineer effective solutions.

Source: Machine learning: Trends, perspectives, and prospects, M. I. Jordan, T. M. Mitchell, Science 17 Jul 2015: Vol. 349, Issue 6245, pp. 255-260, DOI: 10.1126/science.aaa8415

key insights in the above paper

• ML is concerned with the building of computers that improve automatically through experience
• ML lies at the intersection of computer science and statistics and at the core of artificial intelligence and data science
• Recent progress in ML is due to:
  • development of new algorithms and theory
  • ongoing explosion in the availability of online data
  • availability of low-cost computation (through clusters of commodity hardware in the cloud )
• The adoption of data science and ML methods is leading to more evidence-based decision-making across:
  • life sciences (neuroscience, genomics, agriculture, etc. )
  • manufacturing
  • robotics (autonomous vehicle)
  • vision, speech processing, natural language processing
  • education
  • financial modeling
  • policing
  • marketing

US CIA's CTO Gus Hunt's View on Big Data and Data Science

This is recommended viewing for historical insights into Big Data and Data Science skills - thanks to Snowden's Permanent Record for this pointer. At least watch for a few minutes from about 23 minutes to see what Gus Hunt, the then CTO of the American CIA, thinks about the combination of skills needed for Data Science. Watch the whole talk by Gus Hunt titled CIA's Chief Tech Officer on Big Data: We Try to Collect Everything and Hang Onto It Forever if you have 20 minutes or so to get a better framework for this first lecture on the data science process.

Apache Spark actually grew out of Obama era Big Data straetigic grants to UC Berkeley's AMP Lab (the academic origins of Apark and databricks).

The Gus Hunt's distinction proposed here between enumeration versus modeling is mathematically fundamental. The latter is within the realm of classical probabilistic learning theory (eg. Probabilistic Theory of Pattern Recognition, Devroye, Luc; Györfi, László; Lugosi, Gábor, 1996), including, Deep Learning, Reinforcement Learning, etc.), while the former is partly frameable within a wider mathematical setting known as predicting individual sequences (eg. Prediction, Learning and Games, Cesa-Bianchi, N., & Lugosi, G., 2006).

But what is Data Engineering (including Machine Learning Engineering and Operations) and how does it relate to Data Science?

Data Engineering

There are several views on what a data engineer is supposed to do:

Some views are rather narrow and emphasise division of labour between data engineers and data scientists:

• https://www.oreilly.com/ideas/data-engineering-a-quick-and-simple-definition
  • Let's check out what skills a data engineer is expected to have according to the link above.

"Ian Buss, principal solutions architect at Cloudera, notes that data scientists focus on finding new insights from a data set, while data engineers are concerned with the production readiness of that data and all that comes with it: formats, scaling, resilience, security, and more."

What skills do data engineers need? Those “10-30 different big data technologies” Anderson references in “Data engineers vs. data scientists” can fall under numerous areas, such as file formats, > ingestion engines, stream processing, batch processing, batch SQL, data storage, cluster management, transaction databases, web frameworks, data visualizations, and machine learning. And that’s just the tip of the iceberg.

Buss says data engineers should have the following skills and knowledge:

• They need to know Linux and they should be comfortable using the command line.
• They should have experience programming in at least Python or Scala/Java.
• They need to know SQL.
• They need some understanding of distributed systems in general and how they are different from traditional storage and processing systems.
• They need a deep understanding of the ecosystem, including ingestion (e.g. Kafka, Kinesis), processing frameworks (e.g. Spark, Flink) and storage engines (e.g. S3, HDFS, HBase, Kudu). They should know the strengths and weaknesses of each tool and what it's best used for.
• They need to know how to access and process data.

Let's dive deeper into such highly compartmentalised views of data engineers and data scientists and the so-called "machine learning engineers" according the following view:

• https://www.oreilly.com/ideas/data-engineers-vs-data-scientists

Industry will keep evolving, so expect new titles in the future to get new types of jobs done.

The Data Engineering Scientist as "The Middle Way"

Here are some basic axioms that should be self-evident.

• Yes, there are differences in skillsets across humans
  • some humans will be better and have inclinations for engineering and others for pure mathematics by nature and nurture
  • one human cannot easily be a master of everything needed for innovating a new data-based product or service (very very rarely though this happens)
• Skills can be gained by any human who wants to learn to the extent s/he is able to expend time, energy, etc.

For the Scalable Data Engineering Science Process: towards Production-Ready and Productisable Prototyping we need to allow each data engineer to be more of a data scientist and each data scientist to be more of a data engineer, up to each individual's comfort zones in technical and mathematical/conceptual and time-availability planes, but with some minimal expectations of mutual appreciation.

This course is designed to help you take the first minimal steps towards such a data engineering science process.

In the sequel it will become apparent why a team of data engineering scientists with skills across the conventional (2022) spectrum of data engineer versus data scientist is crucial for Production-Ready and Productisable Prototyping, whose outputs include standard AI products today.

A Brief Tour of Data Science

History of Data Analysis and Where Does "Big Data" Come From?

• A Brief History and Timeline of Data Analysis and Big Data
• https://en.wikipedia.org/wiki/Big_data
• https://whatis.techtarget.com/feature/A-history-and-timeline-of-big-data
• Where does Data Come From?
• Some of the sources of big data.
  • online click-streams (a lot of it is recorded but a tiny amount is analyzed):
    • record every click
    • every ad you view
    • every billing event,
    • every transaction, every network message, and every fault.
  • User-generated content (on web and mobile devices):
    • every post that you make on Facebook
    • every picture sent on Instagram
    • every review you write for Yelp or TripAdvisor
    • every tweet you send on Twitter
    • every video that you post to YouTube.
  • Science (for scientific computing):
    • data from various repositories for natural language processing:
      • Wikipedia,
      • the Library of Congress,
      • twitter firehose and google ngrams and digital archives,
    • data from scientific instruments/sensors/computers:
      • the Large Hadron Collider (more data in a year than all the other data sources combined!)
      • genome sequencing data (sequencing cost is dropping much faster than Moore's Law!)
      • output of high-performance computers (super-computers) for data fusion, estimation/prediction and exploratory data analysis
  • Graphs are also an interesting source of big data (network science).
    • social networks (collaborations, followers, fb-friends or other relationships),
    • telecommunication networks,
    • computer networks,
    • road networks
  • machine logs:
    • by servers around the internet (hundreds of millions of machines out there!)
    • internet of things.

Data Science with Cloud Computing and What's Hard about it?

• See Cloud Computing to understand the work-horse for analysing big data at data centers

Cloud computing is the on-demand availability of computer system resources, especially data storage (cloud storage) and computing power, without direct active management by the user. Large clouds often have functions distributed over multiple locations, each location being a data center. Cloud computing relies on sharing of resources to achieve coherence and economies of scale, typically using a "pay-as-you-go" model which can help in reducing capital expenses but may also lead to unexpected operating expenses for unaware users.

• In fact, if you are logged into https://*.databricks.com/* you are computing in the cloud! So the computations are actually running in an instance of the hardware available at a data center like the following:

  • 

• Here is a data center used by CERN in 2010.

  • 

• What's hard about scalable data science in the cloud?

  • To analyse datasets that are big, say more than a few TBs, we need to split the data and put it in several computers that are networked - a typical cloud
  • 
  • However, as the number of computer nodes in such a network increases, the probability of hardware failure or fault (say the hard-disk or memory or CPU or switch breaking down) also increases and can happen while the computation is being performed
  • Therefore for scalable data science, i.e., data science that can scale with the size of the input data by adding more computer nodes, we need fault-tolerant computing and storage framework at the software level to ensure the computations finish even if there are hardware faults.

Here is a recommended light reading on What is "Big Data" -- Understanding the History (18 minutes): - https://towardsdatascience.com/what-is-big-data-understanding-the-history-32078f3b53ce

What should you be able to do at the end of this course?

By following these online interactions in the form of lab/lectures, asking questions, engaging in discussions, doing HOMEWORK assignments and completing the group project, you should be able to:

• Understand the principles of fault-tolerant scalable computing in Spark
  • in-memory and generic DAG extensions of Map-reduce
  • resilient distributed datasets for fault-tolerance
  • skills to process today's big data using state-of-the art techniques in Apache Spark 3.0, in terms of:
    • hands-on coding with realistic datasets
    • an intuitive understanding of the ideas behind the technology and methods
    • pointers to academic papers in the literature, technical blogs and video streams for you to futher your theoretical understanding.
• More concretely, you will be able to:
  • Extract, Transform, Load, Interact, Explore and Analyze Data
  • Build Scalable Machine Learning Pipelines (or help build them) using Distributed Algorithms and Optimization
• How to keep up?
  • This is a fast-changing world.
  • Recent videos around Apache Spark are archived here (these videos are a great way to learn the latest happenings in industrial R&D today!):
    • https://databricks.com/dataaisummit
• What is mathematically stable in the world of 'big data'?
  • There is a growing body of work on the analysis of parallel and distributed algorithms, the work-horse of big data and AI.
  • We will see the core of this in the theoretical material from Reza Zadeh's course on Distributed Algorithms and Optimisation.

Data Science Process under the Algorithms-Machines-Peoples-Planet Framework

Preparatory perusal at some distance, without necessarily associating oneself with any particular "philosophy", in order to intuitively understand the Data Science Process under the Algorithms-Machines-Peoples-Planet (AMPP) Framework, as formal Decision Problems and Decision Procedures (including any AI/ML algorithm) for consideration before Action using Mathematical Decision Theory.

1. Kate Crawford and Vladan Joler, “Anatomy of an AI System: The Amazon Echo As An Anatomical Map of Human Labor, Data and Planetary Resources,” AI Now Institute and Share Lab, (September 7, 2018)
   • Browse anatomyof.ai.
   • View full-scale map as PDF
   • Read Map + Essay as PDF in A3 format (expected time is about 1.5 hours for thorough comprehension)
2. MANIFESTO ON THE FUTURE OF SEEDS. Produced by The International Commission on the Future of Food and Agriculture, 36 pages. (2006). Disseminated as PDF from http://lamastex.org/JOE/ManifestoOnFutureOfSeeds*2006.pdf.
3. The Joint Operating Environment (JOE) limited to the perspective of The United States: 
   • Read at least page 3 on "About this Study" https://www.jcs.mil/Portals/36/Documents/Doctrine/concepts/joe*2008.pdf
     
     • Do a deeper dive if interested at https://www.jcs.mil/Doctrine/Joint-Concepts/JOE/ following through Joint Operating Environment, 2010 and Joint Operating Environment 2035, 14 July 2016
4. Know the doctrine of Mutual Assured Destruction (MAD). Peruse: https://en.wikipedia.org/wiki/Mutual*assured*destruction 
5. Peruse https://www.oneearth.org/our-mission/, and more specifically inform yourself by further perusing
   1. Energy Transition: https://www.oneearth.org/science/energy/
   2. Nature Conservation: https://www.oneearth.org/science/nature/
   3. Regenerative Agriculture: https://www.oneearth.org/science/agriculture/ 

Interactions

When you are involved in a data science process (to "make a living", say) under the AMPP framework, your Algorithms implemented on Machines can have different effects on Peoples (meaning, any living populations of any species, including different human sub-populations, plants, animals, microbes, etc.) and our Planet (soils, climates, oceans, etc.) as long as the Joint Operating Environment is stable to avoid Mutual Assured Destruction.

Discussion 0

• Is it important to be aware of such different effects when building of a data product in some data science process?
  • We will limit the discussions on the concrete matter of "Alexa", AWS's voice assistant.
    • what are your thoughts on https://anatomyof.ai/?
      • view https://youtu.be/uM7gqPnmDDc
      • view https://youtu.be/i_EZwV6wGP0

This is a primer for our industrial guest speakers from https://www.trase.earth/ who will talk about supply chains.

ScaDaMaLe Course site and book

Why Apache Spark?

• Apache Spark: A Unified Engine for Big Data Processing By Matei Zaharia, Reynold S. Xin, Patrick Wendell, Tathagata Das, Michael Armbrust, Ankur Dave, Xiangrui Meng, Josh Rosen, Shivaram Venkataraman, Michael J. Franklin, Ali Ghodsi, Joseph Gonzalez, Scott Shenker, Ion Stoica Communications of the ACM, Vol. 59 No. 11, Pages 56-65 10.1145/2934664

Right-click the above image-link, open in a new tab and watch the video (4 minutes) or read about it in the Communications of the ACM in the frame below or from the link above.

Key Insights from Apache Spark: A Unified Engine for Big Data Processing

• A simple programming model can capture streaming, batch, and interactive workloads and enable new applications that combine them.
• Apache Spark applications range from finance to scientific data processing and combine libraries for SQL, machine learning, and graphs.
• In six years, Apache Spark has grown to 1,000 contributors and thousands of deployments.

Spark 3.0 is the latest version now (20200918) and it should be seen as the latest step in the evolution of tools in the big data ecosystem as summarized in https://towardsdatascience.com/what-is-big-data-understanding-the-history-32078f3b53ce:

Alternatives to Apache Spark

There are several alternatives to Apache Spark, but none of them have the penetration and community of Spark as of 2021.

For real-time streaming operations Apache Flink is competitive. See Apache Flink vs Spark – Will one overtake the other? for a July 2021 comparison. Most scalable data science and engineering problems faced by several major industries in Sweden today are routinely solved using tools in the ecosystem around Apache Spark. Therefore, we will focus on Apache Spark here which still holds the world record for 10TB or 10,000 GB sort by Alibaba cloud in 06/17/2020.

Several alternatives to Apache Spark exist. See the following for verious commercial options: - https://sourceforge.net/software/product/Apache-Spark/alternatives

Read the following for a comparison of three popular frameworks in 2022 for distributed computing:

• https://www.dominodatalab.com/blog/spark-dask-ray-choosing-the-right-framework

Here, we will focus on Apache Spark as it is still a popular framework for distributed computing with a rich ecosystem around it.

The big data problem

Hardware, distributing work, handling failed and slow machines

Let us recall and appreciate the following:

• The Big Data Problem
  • Many routine problems today involve dealing with "big data", operationally, this is a dataset that is larger than a few TBs and thus won't fit into a single commodity computer like a powerful desktop or laptop computer.
• Hardware for Big Data
• The best single commodity computer can not handle big data as it has limited hard-disk and memory
• Thus, we need to break the data up into lots of commodity computers that are networked together via cables to communicate instructions and data between them - this can be thought of as a cloud
• How to distribute work across a cluster of commodity machines?
  • We need a software-level framework for this.
• How to deal with failures or slow machines?
  • We also need a software-level framework for this.

Key Papers

• Key Historical Milestones
  • 1956-1979: Stanford, MIT, CMU, and other universities develop set/list operations in LISP, Prolog, and other languages for parallel processing
  • 2004: READ: Google's MapReduce: Simplified Data Processing on Large Clusters, by Jeffrey Dean and Sanjay Ghemawat
  • 2006: Yahoo!'s Apache Hadoop, originating from the Yahoo!’s Nutch Project, Doug Cutting - wikipedia
  • 2009: Cloud computing with Amazon Web Services Elastic MapReduce, a Hadoop version modified for Amazon Elastic Cloud Computing (EC2) and Amazon Simple Storage System (S3), including support for Apache Hive and Pig.
  • 2010: READ: The Hadoop Distributed File System, by Konstantin Shvachko, Hairong Kuang, Sanjay Radia, and Robert Chansler. IEEE MSST
• Apache Spark Core Papers
  • 2012: READ: Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing, Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, Ankur Dave, Justin Ma, Murphy McCauley, Michael J. Franklin, Scott Shenker and Ion Stoica. NSDI
  • 2016: Apache Spark: A Unified Engine for Big Data Processing By Matei Zaharia, Reynold S. Xin, Patrick Wendell, Tathagata Das, Michael Armbrust, Ankur Dave, Xiangrui Meng, Josh Rosen, Shivaram Venkataraman, Michael J. Franklin, Ali Ghodsi, Joseph Gonzalez, Scott Shenker, Ion Stoica , Communications of the ACM, Vol. 59 No. 11, Pages 56-65, 10.1145/2934664
• A lot has happened since 2016 to improve efficiency of Spark and embed more into the big data ecosystem
  • See Introducing Apache Spark 3.0 | Matei Zaharia and Brooke Wenig | Keynote Spark + AI Summit 2020.
• More research papers on Spark are available from here and we will refer to them in the sequel as needed:
  • https://databricks.com/resources?sftresource_type=research-papers

MapReduce and Apache Spark.

MapReduce as we will see shortly in action is a framework for distributed fault-tolerant computing over a fault-tolerant distributed file-system, such as Google File System or open-source Hadoop for storage.

• Unfortunately, Map Reduce is bounded by Disk I/O and can be slow
  • especially when doing a sequence of MapReduce operations requirinr multiple Disk I/O operations
• Apache Spark can use Memory instead of Disk to speed-up MapReduce Operations
  • Spark Versus MapReduce - the speed-up is orders of magnitude faster
• SUMMARY
  • Spark uses memory instead of disk alone and is thus fater than Hadoop MapReduce
  • Spark's resilience abstraction is by RDD (resilient distributed dataset)
  • RDDs can be recovered upon failures from their lineage graphs, the recipes to make them starting from raw data
  • Spark supports a lot more than MapReduce, including streaming, interactive in-memory querying, etc.
  • Spark demonstrated an unprecedented sort of 1 petabyte (1,000 terabytes) worth of data in 234 minutes running on 190 Amazon EC2 instances (in 2015).
  • Spark expertise corresponds to the highest Median Salary in the US (~ 150K)

ScaDaMaLe Course site and book

Login to databricks

We will use databricks community edition and later on the databricks project shard granted for this course under the databricks university alliance with cloud computing grants from databricks for waived DBU units and AWS.

Please go here for a relaxed and detailed-enough tour (later):

• https://docs.databricks.com/index.html

databricks community edition

First obtain a free Obtain a databricks community edition account at https://community.cloud.databricks.com by following these instructions: https://youtu.be/FH2KDhaFkZg.

Let's get an overview of the databricks managed cloud for processing big data with Apache Spark.

DBC Essentials: What is Databricks Cloud?

DBC Essentials: Shard, Cluster, Notebook and Dashboard

DBC Essentials: Team, State, Collaboration, Elastic Resources in one picture

You Should All Have databricks community edition account by now! and have successfully logged in to it.

Import Course Content Now!

Two Steps:

1. Create a folder named scalable-data-science in your Workspace (NO Typos due to hard-coding of paths in the sequel!)

• Import the following .dbc archives from the following URL into Workspace/scalable-data-science folder you just created:
  • https://github.com/lamastex/scalable-data-science/raw/master/dbcArchives/latest/
  • start with the first file for now and import more as needed:
    • https://github.com/lamastex/scalable-data-science/raw/master/dbcArchives/latest/000_1-sds-3-x-spark.dbc
    • ...

Open-Source Computing Environment

We will mainly use docker for local development on our own infrastructure, such as a laptop.

docker

You can install and learn docker on your laptop; https://docs.docker.com/get-started/.

In a Terminal, after installing docker, type:

$ docker run --rm -it lamastex/dockerdev:latest

The above docker container has already downloaded the needed sources for a very basic Spark developer environment using this Dockerfile which builds on this spark3x.Dockerfile. Having the source for a dockerfile is helpful when one needs to adapt it for the future.

You can test your docker installation by launching the spark-shell inside the docker container, compute 1+1 and :quit from spark-shell and exit from docker.

root@7715366c86a4:~# spark-shell
22/08/18 12:24:26 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://7715366c86a4:4040
Spark context available as 'sc' (master = local[*], app id = local-1660825477089).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.1.2
      /_/
         
Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 1.8.0_312)
Type in expressions to have them evaluated.
Type :help for more information.

scala> 1+1
res0: Int = 2

scala> :quit
root@c6ced3348c97:~# exit

zeppelin

You may also use the open source project https://zeppelin.apache.org/ through the provided zeppelin spark docker container lamastex/zeppelin-spark after successfully installing docker in your laptop.

Several companies use databricks notebooks but several others use zeppelin or jupyter or other notebook servers. Therefore, it is good to be familiar with a couple notebook servers and formats.

ScaDaMaLe Course site and book

Please go here for a relaxed and detailed-enough tour (later):

• https://docs.databricks.com/index.html

Multi-lingual Notebooks

Write Spark code for processing your data in notebooks.

Note that there are several open-sourced notebook servers including:

• zeppelin
• jupyter and its variants
• polynote
• for a list of notebook alternatives to databricks see:
  • https://datasciencenotebook.org/alternatives/databricks

For comparison between pairs of notebook servers see:

• https://datasciencenotebook.org/compare/zeppelin/databricks
• https://datasciencenotebook.org/compare/jupyter/databricks

Here, we are mainly focused on using databricks notebooks due to its effeciently managed engineering layers over AWS (or Azure or GPC public clouds).

This allows us to focus on the concepts and models and get to implement codes with ease.

NOTE: By now you should have already opened this notebook and attached it to a cluster:

• that you started in the Community Edition of databricks
• or that you started in the zeppelin spark docker container similar to lamastex/zeppelin-spark

Databricks Notebook

Next we delve into the mechanics of working with databricks notebooks.

• But many of the details also apply to other notebook environments such as zeppelin with minor differences.

Some Zeppelin caveats

• Since many companies use other notebook servers we will make remarks on minor differences between databricks and zeppelin notebook formats, in case you want to try zeppelin server on your laptop or your own infrastructure.
• Note that one can run Jupyter within Zeppelin. We use Zeppelin as opposed to Jupyter here because Zeppelin comes with a much broader collection of default interpreters for today's big data ecosystem and we wanted to only have one other open-source notebook server to complement learning in the currently (2022) closed-source notebook server of databricks.

Notebooks can be written in Python, Scala, R, or SQL.

• This is a Scala notebook - which is indicated next to the title above by (Scala).
• One can choose the default language of the notebook when it is created.

Creating a new Notebook

• Select Create > Notebook.
• Enter the name of the notebook, the language (Python, Scala, R or SQL) for the notebook, and a cluster to run it on.
• From Left Menu you can go to 'Workspace' and then Click the tiangle on the right side of a folder to open the folder menu and create notebook in the folder.

Cloning a Notebook

• You can clone a notebook to create a copy of it, for example if you want to edit or run an Example notebook like this one.
• Click File > Clone in the notebook context bar above.
• Enter a new name and location for your notebook. If Access Control is enabled, you can only clone to folders that you have Manage permissions on.

Clone Or Import This Notebook

• From the File menu at the top left of this notebook, choose Clone or click Import Notebook on the top right. This will allow you to interactively execute code cells as you proceed through the notebook.

• Enter a name and a desired location for your cloned notebook (i.e. Perhaps clone to your own user directory or the "Shared" directory.)

• Navigate to the location you selected (e.g. click Menu > Workspace > Your cloned location)

Attach the Notebook to a cluster

• A Cluster is a group of machines which can run commands in cells.
• Check the upper left corner of your notebook to see if it is Attached or Detached.
• If Detached, click on the right arrow and select a cluster to attach your notebook to.
  • If there is no running cluster, create one.

Deep-dive into databricks notebooks

Let's take a deeper dive into a databricks notebook next.

Cells are units that make up notebooks

Cells each have a type - including scala, python, sql, R, markdown, filesystem, and shell.

• While cells default to the type of the Notebook, other cell types are supported as well.
• This cell is in markdown and is used for documentation. Markdown is a simple text formatting syntax.

Beware of minor differences across notebook formats

• Different notebook formats use different tags for the same language interpreter. But they are quite easy to remember.
  • For example in zeppelin %spark and %pyspark are used instead of databricks' %scala and %python to denote spark and pyspark cells.
  • However %md and %sh are used for markdown and shell in both notebook formats. ***

Create and Edit a New Markdown Cell in this Notebook

NOTE: You will be writing your group project report as databricks notebooks, therefore it is important to use markdown effectively.

• When you mouse between cells, a + sign will pop up in the center that you can click on to create a new cell.

• Type %md Hello, world! into your new cell (%md indicates the cell is markdown).

• Click out of the cell or double-click to see the cell contents update.

Running a cell in your notebook.

• Press Shift+Enter when in the cell to run it and proceed to the next cell.
  • The cells contents should update.
• NOTE: Cells are not automatically run each time you open it.
  • Instead, Previous results from running a cell are saved and displayed.
• Alternately, press Ctrl+Enter when in a cell to run it, but not proceed to the next cell.

You Try Now! Just double-click the cell below, modify the text following %md and press Ctrl+Enter to evaluate it and see it's mark-down'd output.

> %md Hello, world!

Hello, world!

Markdown Cell Tips

• To change a non-markdown cell to markdown, add %md to very start of the cell.
• After updating the contents of a markdown cell, click out of the cell to update the formatted contents of a markdown cell.
• To edit an existing markdown cell, doubleclick the cell.

Learn more about markdown:

• https://guides.github.com/features/mastering-markdown/

Note that there are flavours or minor variants and enhancements of markdown, including those specific to databricks, github, pandoc, etc.

It will be future-proof to remain in the syntactic zone of pure markdown (at the intersection of various flavours) as much as possible and go with pandoc-compatible style if choices are necessary. ***

Run a Scala Cell

• Run the following scala cell.
• Note: There is no need for any special indicator (such as %md) necessary to create a Scala cell in a Scala notebook.
• You know it is a scala notebook because of the (Scala) appended to the name of this notebook.
• Make sure the cell contents updates before moving on.
• Press Shift+Enter when in the cell to run it and proceed to the next cell.
  • The cells contents should update.
  • Alternately, press Ctrl+Enter when in a cell to run it, but not proceed to the next cell.
• characters following // are comments in scala.
• RECALL: In zeppelin %spark is used instead of %scala. ***

Spark is written in Scala, but ...

For this reason Scala will be the primary language for this course is Scala.

However, let us use the best language for the job! as each cell can be written in a specific language in the same notebook. Such multi-lingual notebooks are the norm in any realistic data science process today!

The beginning of each cells has a language type if it is not the default language of the notebook. Such cell-specific language types include the following with the prefix %:

• %scala for Spark Scala,

  • %spark in zeppelin

• %py for PySpark,

  • %pyspark in zeppelin

• %r for R,

• %sql for SQL,

• %sh for BASH SHELL and

• %md for markdown.

• While cells default to the language type of the Notebook (scala, python, r or sql), other cell types are supported as well in a cell-specific manner.

• For example, Python Notebooks can contain python, sql, markdown, and even scala cells. This lets you write notebooks that do use multiple languages.

• This cell is in markdown as it begins with %mdand is used for documentation purposes.

Thus, all language-typed cells can be created in any notebook, regardless of the the default language of the notebook itself, provided interpreter support exists for the languages.

Cross-language cells can be used to mix commands from other languages.

Examples:

Command line cells can be used to work with local files on the Spark driver node. * Start a cell with %sh to run a command line command

Notebooks can be run from other notebooks using %run

This is commonly used to import functions you defined in other notebooks.

• Syntax in databricks: %run /full/path/to/notebook
• Syntax in zeppelin: z.runNote("id")

Displaying in notebooks

Notebooks are great for displaying data in the input cell immediately in the result or output cell. This is often an effective web-based REPL environment to prototype for the data science process.

To display a dataframe or tabular data named myDataFrame:

• Notebook-agnostic syntax in Spark for textual display: myDataFrame.show()
• Syntax in databricks: display(myDataFrame)
• Syntax in zeppelin: z.show(myDataFrame)

One can also display more sophisticated outputs involving HTML and D3 using display_HTML in databricks or %html in zeppelin. See respective documentation for details. Zeppelin is extremely flexible via helium when it comes to creating interactive visualisations of results from a data science process pipeline.

Further Pointers

Here are some useful links to bookmark as you will need to use them for Reference.

These links provide a relaxed and detailed-enough tour (that you may need to take later):

• databricks
  • https://docs.databricks.com/index.html
• zeppelin (if you want an open-source notebook server deployed on your own computing infrastructure)
  • https://zeppelin.apache.org/docs/0.10.1/

ScaDaMaLe Course site and book

Load Datasets into Databricks

Simply, hit run Runa all cells in this Notebook to load all the core datasets we will be using in the first couple of modules in this course.

It can take about 2-3 minutes once your cluster starts.

Note This notebook should be skipped in docker-compose zeppelin instance of the course.

Now you have loaded the core datasets for the course.

ScaDaMaLe Course site and book

Scala Crash Course

Here we take a minimalist approach to learning just enough Scala, the language that Apache Spark is written in, to be able to use Spark effectively.

In the sequel we can learn more Scala concepts as they arise. This learning can be done by chasing the pointers in this crash course for a detailed deeper dive on your own time.

There are two basic ways in which we can learn Scala:

1. Learn Scala in a notebook environment

For convenience we use databricks Scala notebooks like this one here.

You can learn Scala locally on your own computer using Scala REPL (and Spark using Spark-Shell).

2. Learn Scala in your own computer

The most easy way to get Scala locally is through sbt, the Scala Build Tool. You can also use an IDE that integrates sbt.

See: https://docs.scala-lang.org/getting-started/index.html to set up Scala in your own computer.

Software Engineering NOTE: You can use docker pull lamastex/dockerdev:spark3x and use the docker container for local development based on minimal pointers here: https://github.com/lamastex/dockerDev. Being able to modularise and reuse codes modularly requires the building of libraries which inturn requires building them locally with tools such as sbt or mvn for instance. You can also set-up modern IDEs like https://code.visualstudio.com/ or https://www.jetbrains.com/idea/, etc. for this.

Scala Resources

You will not be learning scala systematically and thoroughly in this course. You will learn to use Scala by doing various Spark jobs.

If you are interested in learning scala properly, then there are various resources, including:

• scala-lang.org is the core Scala resource. Bookmark the following three links:
  • tour-of-scala - Bite-sized introductions to core language features.
    • we will go through the tour in a hurry now as some Scala familiarity is needed immediately.
  • scala-book - An online book introducing the main language features
    • you are expected to use this resource to figure out Scala as needed.
  • scala-cheatsheet - A handy cheatsheet covering the basics of Scala syntax.
  • visual-scala-reference - This guide collects some of the most common functions of the Scala Programming Language and explain them conceptual and graphically in a simple way.
• Online Resources, including:
  • courseera: Functional Programming Principles in Scala
• Books
  • Programming in Scala, 1st Edition, Free Online Reading

The main sources for the following content are (you are encouraged to read them for more background):

• Martin Oderski's Scala by example
• Scala crash course by Holden Karau
• Darren's brief introduction to scala and breeze for statistical computing

What is Scala?

"Scala smoothly integrates object-oriented and functional programming. It is designed to express common programming patterns in a concise, elegant, and type-safe way." by Matrin Odersky.

• High-level language for the Java Virtual Machine (JVM)
• Object oriented + functional programming
• Statically typed
• Comparable in speed to Java
• Type inference saves us from having to write explicit types most of the time Interoperates with Java
• Can use any Java class (inherit from, etc.)
• Can be called from Java code

See a quick tour here:

• https://docs.scala-lang.org/tour/tour-of-scala.html

Why Scala?

• Spark was originally written in Scala, which allows concise function syntax and interactive use
• Spark APIs for other languages include:
  • Java API for standalone use
  • Python API added to reach a wider user community of programmes
  • R API added more recently to reach a wider community of data analyststs
  • Unfortunately, Python and R APIs are generally behind Spark's native Scala (for eg. GraphX is only available in Scala currently and datasets are only available in Scala as of 20200918).
• See Darren Wilkinson's 11 reasons for scala as a platform for statistical computing and data science. It is embedded in-place below for your convenience.

Scala Versus Python for Apache Spark

Read: https://www.projectpro.io/article/scala-vs-python-for-apache-spark/213 for an interesting overview.

Learn Scala in Notebook Environment

Run a Scala Cell

• Run the following scala cell.
• Note: There is no need for any special indicator (such as %md) necessary to create a Scala cell in a Scala notebook.
• You know it is a scala notebook because of the (Scala) appended to the name of this notebook.
• Make sure the cell contents updates before moving on.
• Press Shift+Enter when in the cell to run it and proceed to the next cell.
  • The cells contents should update.
  • Alternately, press Ctrl+Enter when in a cell to run it, but not proceed to the next cell.
• characters following // are comments in scala. ***

See Scala as a platform for statistical computing and data science.

Let's get our hands dirty in Scala

We will go through the following programming concepts and tasks by building on https://docs.scala-lang.org/tour/basics.html.

• Scala Types
• Expressions and Printing
• Naming and Assignments
• Functions and Methods in Scala
• Classes and Case Classes
• Methods and Tab-completion
• Objects and Traits
• Collections in Scala and Type Hierarchy
• Functional Programming and MapReduce
• Lazy Evaluations and Recursions

Remark: You need to take a computer science course (from CourseEra, for example) to properly learn Scala. Here, we will learn to use Scala by example to accomplish our data science tasks at hand. You can learn more Scala as needed from various sources pointed out above in Scala Resources.

Scala Types

In Scala, all values have a type, including numerical values and functions. The diagram below illustrates a subset of the type hierarchy.

For now, notice some common types we will be usinf including Int, String, Double, Unit, Boolean, List, etc. For more details see https://docs.scala-lang.org/tour/unified-types.html. We will return to this at the end of the notebook after seeing a brief tour of Scala now.

Expressions

Expressions are computable statements such as the 1+1 we have seen before.

We can print the output of a computed or evaluated expressions as a line using println:

Naming and Assignments

value and variable as val and var

You can name the results of expressions using keywords val and var.

Let us assign the integer value 5 to x as follows:

x is a named result and it is a value since we used the keyword val when naming it.

Scala is statically typed, but it uses built-in type inference machinery to automatically figure out that x is an integer or Int type as follows. Let's declare a value x to be Int 5 next without explictly using Int.

Let's declare x as a Double or double-precision floating-point type using decimal such as 5.0 (a digit has to follow the decimal point!)

Alternatively, we can assign x as a Double explicitly. Note that the decimal point is not needed in this case due to explicit typing as Double.

Next note that labels need to be declared on first use. We have declared x to be a val which is short for value. This makes x immutable (cannot be changed).

Thus, x cannot be just re-assigned, as the following code illustrates in the resulting error: ... error: reassignment to val.

Scala allows declaration of mutable variables as well using var, as follows:

Blocks

Just combine expressions by surrounding them with { and } called a block.

Functions

Functions are expressions that have parameters. A function takes arguments as input and returns expressions as output.

A function can be nameless or anonymous and simply return an output from a given input. For example, the following annymous function returns the square of the input integer.

On the left of => is a list of parameters with name and type. On the right is an expression involving the parameters.

You can also name functions:

A function can have no parameters:

A function can have more than one parameter:

Methods

Methods are very similar to functions, but a few key differences exist.

Methods use the def keyword followed by a name, parameter list(s), a return type, and a body.

Note that the return type Int is specified after the parameter list and a :.

Scala has a return keyword but it is rarely used as the expression in the last line of the multi-line block is the method's return value.

A method can have output expressions involving multiple parameter lists:

A method can have no parameter lists at all:

Classes

The class keyword followed by the name and constructor parameters is used to define a class.

• The return type of the method printVolume is Unit.
• When the return type is Unit it indicates that there is nothing meaningful to return, similar to void in Java and C, but with a difference.
• Because every Scala expression must have some value, there is actually a singleton value of type Unit, written () and carrying no information.

We can make an instance of the class with the new keyword.

And call the method on the instance.

Our named instance my1Cube of the Box class is immutable due to val.

You can have mutable instances of the class using var.

See https://docs.scala-lang.org/tour/classes.html for more details as needed.

Case Classes

Scala has a special type of class called a case class that can be defined with the case class keyword.

Unlike classes, whose instances are compared by reference, instances of case classes are immutable by default and compared by value. This makes them useful for defining rows of typed values in Spark.

Case classes can be instantiated without the new keyword.

Instances of case classes are compared by value and not by reference.

By contrast, instances of classes are compared by reference.

More about case classes here: https://docs.scala-lang.org/tour/case-classes.html.

Methods and Tab-completion

Many methods of a class can be accessed by ..

You can place the cursor after . following a declared object and find out the methods available for it as shown in the image below.

You Try doing this next.

For example,

• scroll down to contains and double-click on it.
• This should lead to s.contains in your cell.
• Now add an argument String to see if s contains the argument, for example, try:
  • s.contains("f")
  • s.contains("") and
  • s.contains("i")

Objects

Objects are single instances of their own definitions using the object keyword. You can think of them as singletons of their own classes.

You can access an object through its name:

For details see https://docs.scala-lang.org/tour/singleton-objects.html

Traits

Traits are abstract data types containing certain fields and methods. They can be defined using the trait keyword.

In Scala inheritance, a class can only extend one other class, but it can extend multiple traits.

Traits can have default implementations also.

You can extend traits with the extends keyword and override an implementation with the override keyword:

Instantiate the classes.

Call the greet method in each case.

A class can also be made to extend multiple traits.

For more details see: https://docs.scala-lang.org/tour/traits.html.

Main Method

The main method is the entry point of a Scala program.

The Java Virtual Machine requires a main method, named main, that takes an array of strings as its only argument.

Using an object, you can define the main method as follows:

What I try not do while learning a new language?

1. I don't immediately try to ask questions like: how can I do this particular variation of some small thing I just learned so I can use patterns I am used to from another language I am hooked-on right now?
2. first go through the detailed Scala Tour on your own and then through the 50 odd lessons in the Scala Book
3. then return to 1. and ask detailed cross-language comparison questions by diving deep as needed with the source and scala docs as needed (google or duck-duck-go search!).

ScaDaMaLe Course site and book

Scala Crash Course Continued

Recall!

Scala Resources

You will not be learning scala systematically and thoroughly in this course. You will learn to use Scala by doing various Spark jobs.

If you are interested in learning scala properly, then there are various resources, including:

• scala-lang.org is the core Scala resource. Bookmark the following three links:
  • tour-of-scala - Bite-sized introductions to core language features.
    • we will go through the tour in a hurry now as some Scala familiarity is needed immediately.
  • scala-book - An online book introducing the main language features
    • you are expected to use this resource to figure out Scala as needed.
  • scala-cheatsheet - A handy cheatsheet covering the basics of Scala syntax.
  • visual-scala-reference - This guide collects some of the most common functions of the Scala Programming Language and explain them conceptual and graphically in a simple way.
• Online Resources, including:
  • courseera: Functional Programming Principles in Scala
• Books
  • Programming in Scala, 1st Edition, Free Online Reading

The main sources for the following content are (you are encouraged to read them for more background):

• Martin Oderski's Scala by example
• Scala crash course by Holden Karau
• Darren's brief introduction to scala and breeze for statistical computing

Let's continue getting our hands dirty in Scala

We will go through the remaining programming concepts and tasks by building on https://docs.scala-lang.org/tour/basics.html.

• Scala Types
• Expressions and Printing
• Naming and Assignments
• Functions and Methods in Scala
• Classes and Case Classes
• Methods and Tab-completion
• Objects and Traits
• Collections in Scala and Type Hierarchy
• Functional Programming and MapReduce
• Lazy Evaluations and Recursions

Remark: You need to take a computer science course (from CourseEra, for example) to properly learn Scala. Here, we will learn to use Scala by example to accomplish our data science tasks at hand. You can learn more Scala as needed from various sources pointed out above in Scala Resources.

Scala Type Hierarchy

In Scala, all values have a type, including numerical values and functions. The diagram below illustrates a subset of the type hierarchy.

For now, notice some common types we will be usinf including Int, String, Double, Unit, Boolean, List, etc.

Let us take a closer look at Scala Type Hierarchy now here:

• https://docs.scala-lang.org/tour/unified-types.html.

Scala Collections

Familiarize yourself with the main Scala collections classes here:

• https://docs.scala-lang.org/overviews/scala-book/collections-101.html

List

Lists are one of the most basic data structures.

There are several other Scala collections and we will introduce them as needed. The other most common ones are Vector, Array and Seq and the ArrayBuffer.

For details on list see: - https://docs.scala-lang.org/overviews/scala-book/list-class.html

Vectors

The Vector class is an indexed, immutable sequence. The “indexed” part of the description means that you can access Vector elements very rapidly by their index value, such as accessing listOfPeople(999999).

In general, except for the difference that Vector is indexed and List is not, the two classes work the same, so we’ll run through these examples quickly.

For details see: - https://docs.scala-lang.org/overviews/scala-book/vector-class.html

Arrays, Sequences and Tuples

See https://www.scala-lang.org/api/current/scala/collection/index.html for docs.

A tuple is a neat class that gives you a simple way to store heterogeneous (different) items in the same container. We will use tuples for key-value pairs in Spark.

See https://docs.scala-lang.org/overviews/scala-book/tuples.html

Functional Programming and MapReduce

"Functional programming is a style of programming that emphasizes writing applications using only pure functions and immutable values. As Alvin Alexander wrote in Functional Programming, Simplified, rather than using that description, it can be helpful to say that functional programmers have an extremely strong desire to see their code as math — to see the combination of their functions as a series of algebraic equations. In that regard, you could say that functional programmers like to think of themselves as mathematicians. That’s the driving desire that leads them to use only pure functions and immutable values, because that’s what you use in algebra and other forms of math."

See https://docs.scala-lang.org/overviews/scala-book/functional-programming.html for short lessons in functional programming.

We will apply functions for processing elements of a scala collection to quickly demonstrate functional programming.

Five ways of adding 1

The first four use anonymous functions and the last one uses a named method.

1. explicit version:

(x: Int) => x + 1  

1. type-inferred more intuitive version:

x => x + 1   

1. placeholder syntax (each argument must be used exactly once):

_ + 1 

1. type-inferred more intuitive version with code-block for larger function body:

x => { 
      // body is a block of code
      val integerToAdd = 1
      x + integerToAdd
}

1. as methods using def:

def addOne(x: Int): Int = x + 1

Now, let's do some functional programming over scala collection (List) using some of their methods: map, filter and reduce. In the end we will write our first mapReduce program!

For more details see:

• https://docs.scala-lang.org/overviews/scala-book/collections-methods.html

map

See: https://superruzafa.github.io/visual-scala-reference/map/

trait Collection[A] {
  def map[B](f: (A) => B): Collection[B]
}

map creates a collection using as elements the results obtained from applying the function f to each element of this collection.

filter

See: https://superruzafa.github.io/visual-scala-reference/filter/

trait Collection[A] {
  def filter(p: (A) => Boolean): Collection[A]
}

filter creates a collection with those elements that satisfy the predicate p and discarding the rest.

reduce

See: https://superruzafa.github.io/visual-scala-reference/reduce/

trait Collection[A] {
  def reduce(op: (A, A) => A): A
}

reduce applies the binary operator op to pairs of elements in this collection until the final result is calculated.

Let's combine map and reduce programs above to find the sum of after 10 has been added to every element of the original List lst as follows:

Exercise in Functional Programming

You should spend an hour or so going through the Functional Programming Section of the Scala Book:

• https://docs.scala-lang.org/overviews/scala-book/functional-programming.html

Scala lets you write code in an object-oriented programming (OOP) style, a functional programming (FP) style, and even in a hybrid style, using both approaches in combination. This book assumes that you’re coming to Scala from an OOP language like Java, C++, or C#, so outside of covering Scala classes, there aren’t any special sections about OOP in this book. But because the FP style is still relatively new to many developers, we’ll provide a brief introduction to Scala’s support for FP in the next several lessons.

Functional programming is a style of programming that emphasizes writing applications using only pure functions and immutable values. As Alvin Alexander wrote in Functional Programming, Simplified, rather than using that description, it can be helpful to say that functional programmers have an extremely strong desire to see their code as math — to see the combination of their functions as a series of algebraic equations. In that regard, you could say that functional programmers like to think of themselves as mathematicians. That’s the driving desire that leads them to use only pure functions and immutable values, because that’s what you use in algebra and other forms of math.

Functional programming is a large topic, and there’s no simple way to condense the entire topic into this little book, but in the following lessons we’ll give you a taste of FP, and show some of the tools Scala provides for developers to write functional code.

There are lots of methods in Scala Collections. And much more in this scalable language. See for example http://docs.scala-lang.org/cheatsheets/index.html.

Lazy Evaluation

Another powerful programming concept we will need is lazy evaluation -- a form of delayed evaluation. So the value of an expression that is lazily evaluated is only available when it is actually needed.

This is to be contrasted with eager evaluation that we have seen so far -- an expression is immediately evaluated.

Let's demonstrate lazy evaluation using a getTime method and the keyword lazy.

See https://www.scala-exercises.org/scalatutorial/lazyevaluation for more details including the following example with StringBuilder.

What should builder.result() be after the following arithmetic expression involving x,y and z is evaluated?

Lazy Evaluation Exercise - You try Now!

Understand why the output above is what it is!

• Why is z different in its appearance in the final builder string when compared to x and y as we evaluated?

z + y + x + z + y + x

Why Lazy?

Imagine a more complex expression involving the evaluation of millions of values. Lazy evaluation will allow us to actually compute with big data when it may become impossible to hold all the values in memory. This is exactly what Apache Spark does as we will see.

Recursions

Recursion is a powerful framework when a function calls another function, including itself, until some terminal condition is reached.

Here we want to distinguish between two ways of implementing a recursion using a simple example of factorial.

Recall that for any natural number $n$, its factorial is denoted and defined as follows:

$$n! := n \times (n-1) \times (n-2) \times \cdots \times 2 \times 1$$

which has the following recursive expression:

$$n! = n*(n-1)! , , \qquad 0! = 1$$

Let us implement it using two approaches: a naive approach that can run out of memory and another tail-recursive approach that uses constant memory. Read https://www.scala-exercises.org/scalatutorial/tailrecursion for details.

When factorialNaive(4) was evaluated above the following steps were actually done:

factorial(4)
if (4 == 0) 1 else 4 * factorial(4 - 1)
4 * factorial(3)
4 * (3 * factorial(2))
4 * (3 * (2 * factorial(1)))
4 * (3 * (2 * (1 * factorial(0)))
4 * (3 * (2 * (1 * 1)))
24

Notice how we add one more element to our expression at each recursive call. Our expressions becomes bigger and bigger until we end by reducing it to the final value. So the final expression given by a directed acyclic graph (DAG) of the pairwise multiplications given by the right-branching binary tree, whose leaves are input integers and internal nodes are the bianry * operator, can get very large when the input n is large.

Tail recursion is a sophisticated way of implementing certain recursions so that memory requirements can be kept constant, as opposed to naive recursions.

Tail Recursion

That difference in the rewriting rules actually translates directly to a difference in the actual execution on a computer. In fact, it turns out that if you have a recursive function that calls itself as its last action, then you can reuse the stack frame of that function. This is called tail recursion.

And by applying that trick, a tail recursive function can execute in constant stack space, so it's really just another formulation of an iterative process. We could say a tail recursive function is the functional form of a loop, and it executes just as efficiently as a loop.

Implementation of tail recursion in the Exercise below uses Scala annotation, which is a way to associate meta-information with definitions. In our case, the annotation @tailrec ensures that a method is indeed tail-recursive. See the last link to understand how memory requirements can be kept constant in tail recursions.

We mainly want you to know that tail recursions are an important functional programming concept.

Tail Recursion Exercise - You Try Now

• Uncomment the next three cells
• Replace ??? in the next cell with the correct values to make this a tail recursion for factorial.

Functional Programming is a vast subject and we are merely covering the fewest core ideas to get started with Apache Spark asap.

We will return to more concepts as we need them in the sequel.

ScaDaMaLe Course site and book

Introduction to Spark

Spark Essentials: RDDs, Transformations and Actions

• This introductory notebook describes how to get started running Spark (Scala) code in Notebooks.
• Working with Spark's Resilient Distributed Datasets (RDDs)
  • creating RDDs
  • performing basic transformations on RDDs
  • performing basic actions on RDDs

RECOLLECT from 001_WhySpark notebook that Spark does fault-tolerant, distributed, in-memory computing

THEORY CAVEAT This module is focused on getting you to quickly write Spark programs with a high-level appreciation of the underlying concepts.

In the last module, we will spend more time on analyzing the core algorithms in parallel and distributed setting of a typical Spark cluster today -- where several multi-core parallel computers (Spark workers) are networked together to provide a fault-tolerant distributed computing platform.

Spark Cluster Overview:

Driver Program, Cluster Manager and Worker Nodes

The driver does the following:

1. connects to a cluster manager to allocate resources across applications

• acquire executors on cluster nodes
  • executor processs run compute tasks and cache data in memory or disk on a worker node
• sends application (user program built on Spark) to the executors
• sends tasks for the executors to run
  • task is a unit of work that will be sent to one executor

See http://spark.apache.org/docs/latest/cluster-overview.html for an overview of the spark cluster.

The Abstraction of Resilient Distributed Dataset (RDD)

RDD is a fault-tolerant collection of elements that can be operated on in parallel.

Key Points to Note

• Resilient distributed datasets (RDDs) are the primary abstraction in Spark.
• RDDs are immutable once created:
  • can transform it.
  • can perform actions on it.
  • but cannot change an RDD once you construct it.
• Spark tracks each RDD's lineage information or recipe to enable its efficient recomputation if a machine fails.
• RDDs enable operations on collections of elements in parallel.
• We can construct RDDs by:
  • parallelizing Scala collections such as lists or arrays
  • by transforming an existing RDD,
  • from files in distributed file systems such as (HDFS, S3, etc.).
• We can specify the number of partitions for an RDD
• The more partitions in an RDD, the more opportunities for parallelism
• There are two types of operations you can perform on an RDD:
  • transformations (are lazily evaluated)
    • map
    • flatMap
    • filter
    • distinct
    • ...
  • actions (actual evaluation happens)
    • count
    • reduce
    • take
    • collect
    • takeOrdered
    • ...
• Spark transformations enable us to create new RDDs from an existing RDD.
• RDD transformations are lazy evaluations (results are not computed right away)
• Spark remembers the set of transformations that are applied to a base data set (this is the lineage graph of RDD)
• The allows Spark to automatically recover RDDs from failures and slow workers.
• The lineage graph is a recipe for creating a result and it can be optimized before execution.
• A transformed RDD is executed only when an action runs on it.
• You can also persist, or cache, RDDs in memory or on disk (this speeds up iterative ML algorithms that transforms the initial RDD iteratively).
• Here is a great reference URL for programming guides for Spark that one should try to cover first
  • http://spark.apache.org/docs/latest/programming-guide.html.
  • and specifically for RDDs: https://spark.apache.org/docs/latest/rdd-programming-guide.html

Let's get our hands dirty in Spark!

DO NOW!

There is a peer-reviewed Assignment where you will dig deeper into Spark transformations and actions.

Let us look at the legend and overview of the visual RDD Api by doing the following first:

Running Spark

The variable sc allows you to access a Spark Context to run your Spark programs. Recall SparkContext is in the Driver Program.

**NOTE: Do not create the sc variable - it is already initialized for you in spark-shell REPL, that includes notebook environments like databricks, Jupyter, zeppelin, etc. **

We will do the following next:

1. Create an RDD using sc.parallelize

• Perform the collect action on the RDD and find the number of partitions it is made of using getNumPartitions action
• Perform the take action on the RDD
• Transform the RDD by map to make another RDD
• Transform the RDD by filter to make another RDD
• Perform the reduce action on the RDD
• Transform the RDD by flatMap to make another RDD
• Create a Pair RDD
• Perform some transformations on a Pair RDD
• Where in the cluster is your computation running?
• Shipping Closures, Broadcast Variables and Accumulator Variables
• Spark Essentials: Summary
• HOMEWORK
• Importing Standard Scala and Java libraries

Entry Point

Now we are ready to start programming in Spark!

Our entry point for Spark applications is the class SparkSession. An instance of this object is already instantiated for us which can be easily demonstrated by running the next cell

We will need these docs!

• RDD Scala Docs
• Dataset Scala Docs
• https://spark.apache.org/docs/3.0.1/api/scala/index.html you can simply search for other Spark classes, methods, etc here

NOTE that since Spark 2.0 SparkSession is a replacement for the other entry points: * SparkContext, available in our notebook as sc. * SQLContext, or more specifically its subclass HiveContext, available in our notebook as sqlContext.

We will be using the pre-made SparkContext sc when learning about RDDs.

1. Create an RDD using sc.parallelize

First, let us create an RDD of three elements (of integer type Int) from a Scala Seq (or List or Array) with two partitions by using the parallelize method of the available Spark Context sc as follows:

2. Perform the collect action on the RDD and find the number of partitions in it using getNumPartitions action

No action has been taken by sc.parallelize above. To see what is "cooked" by the recipe for RDD x we need to take an action.

The simplest is the collect action which returns all of the elements of the RDD as an Array to the driver program and displays it.

So you have to make sure that all of that data will fit in the driver program if you call collect action!

Let us look at the collect action in detail and return here to try out the example codes.

Let us perform a collect action on RDD x as follows:

CAUTION: collect can crash the driver when called upon an RDD with massively many elements. So, it is better to use other diplaying actions like take or takeOrdered as follows:

Let us look at the getNumPartitions action in detail and return here to try out the example codes.

We can see which elements of the RDD are in which parition by calling glom() before collect().

glom() flattens elements of the same partition into an Array.

Thus from the output above, Array[Array[Int]] = Array(Array(1), Array(2, 3)), we know that 1 is in one partition while 2 and 3 are in another partition.

You Try!

Crate an RDD x with three elements, 1,2,3, and this time do not specifiy the number of partitions. Then the default number of partitions will be used. Find out what this is for the cluster you are attached to.

The default number of partitions for an RDD depends on the cluster this notebook is attached to among others - see programming-guide.

3. Perform the take action on the RDD

The .take(n) action returns an array with the first n elements of the RDD.

You Try!

Fill in the parenthes ( ) below in order to take just one element from RDD x.

4. Transform the RDD by map to make another RDD

The map transformation returns a new RDD that's formed by passing each element of the source RDD through a function (closure). The closure is automatically passed on to the workers for evaluation (when an action is called later).

Let us look at the map transformation in detail and return here to try out the example codes.

5. Transform the RDD by filter to make another RDD

The filter transformation returns a new RDD that's formed by selecting those elements of the source RDD on which the function returns true.

Let us look at the filter transformation in detail and return here to try out the example codes.

6. Perform the reduce action on the RDD

Reduce aggregates a data set element using a function (closure). This function takes two arguments and returns one and can often be seen as a binary operator. This operator has to be commutative and associative so that it can be computed correctly in parallel (where we have little control over the order of the operations!).

Let us look at the reduce action in detail and return here to try out the example codes.

7. Transform an RDD by flatMap to make another RDD

flatMap is similar to map but each element from input RDD can be mapped to zero or more output elements. Therefore your function should return a sequential collection such as an Array rather than a single element as shown below.

Let us look at the flatMap transformation in detail and return here to try out the example codes.

8. Create a Pair RDD

Let's next work with RDD of (key,value) pairs called a Pair RDD or Key-Value RDD.

Let's make a Pair RDD called wordCountPairRDD that is made of (key,value) pairs with key=word and value=1 in order to encode each occurrence of each word in the RDD words, as follows:

Wide Transformations and Shuffles

So far we have seen transformations that are narrow -- with no data transfer between partitions. Think of map.

ReduceByKey and GroupByKey are wide transformations as data has to be shuffled across the partitions in different executors -- this is generally very expensive operation.

READ the Background about Shuffles in the programming guide below.

In Spark, data is generally not distributed across partitions to be in the necessary place for a specific operation. During computations, a single task will operate on a single partition - thus, to organize all the data for a single reduceByKey reduce task to execute, Spark needs to perform an all-to-all operation. It must read from all partitions to find all the values for all keys, and then bring together values across partitions to compute the final result for each key - this is called the shuffle

READ the Performance Impact about Shuffles in the programming guide below.

The Shuffle is an expensive operation since it involves disk I/O, data serialization, and network I/O. To organize data for the shuffle, Spark generates sets of tasks - map tasks to organize the data, and a set of reduce tasks to aggregate it. This nomenclature comes from MapReduce and does not directly relate to Spark’s map and reduce operations.

Internally, results from individual map tasks are kept in memory until they can’t fit. Then, these are sorted based on the target partition and written to a single file. On the reduce side, tasks read the relevant sorted blocks.

https://spark.apache.org/docs/latest/rdd-programming-guide.html#shuffle-operations

9. Perform some transformations on a Pair RDD

Let's next work with RDD of (key,value) pairs called a Pair RDD or Key-Value RDD.

Now some of the Key-Value transformations that we could perform include the following.

• reduceByKey transformation
  • which takes an RDD and returns a new RDD of key-value pairs, such that:
    • the values for each key are aggregated using the given reduced function
    • and the reduce function has to be of the type that takes two values and returns one value.
• sortByKey transformation
  • this returns a new RDD of key-value pairs that's sorted by keys in ascending order
• groupByKey transformation
  • this returns a new RDD consisting of key and iterable-valued pairs.

Let's see some concrete examples next.

Now, let us do just the crucial steps and avoid collecting intermediate RDDs (something we should avoid for large datasets anyways, as they may not fit in the driver program).

You Try!

You try evaluating sortByKey() which will make a new RDD that consists of the elements of the original pair RDD that are sorted by Keys.

The next key value transformation we will see is groupByKey

When we apply the groupByKey transformation to wordCountPairRDD we end up with a new RDD that contains two elements. The first element is the tuple b and an iterable CompactBuffer(1,1,1,1,1) obtained by grouping the value 1 for each of the five key value pairs (b,1). Similarly the second element is the key a and an iterable CompactBuffer(1,1,1,1,1,1) obtained by grouping the value 1 for each of the six key value pairs (a,1).

CAUTION: groupByKey can cause a large amount of data movement across the network. It also can create very large iterables at a worker. Imagine you have an RDD where you have 1 billion pairs that have the key a. All of the values will have to fit in a single worker if you use group by key. So instead of a group by key, consider using reduced by key.

10. Understanding Closures - Where in the cluster is your computation running?

One of the harder things about Spark is understanding the scope and life cycle of variables and methods when executing code across a cluster. RDD operations that modify variables outside of their scope can be a frequent source of confusion. In the example below we’ll look at code that uses foreach() to increment a counter, but similar issues can occur for other operations as well.

https://spark.apache.org/docs/latest/rdd-programming-guide.html#understanding-closures-

From RDD programming guide:

The behavior of the above code is undefined, and may not work as intended. To execute jobs, Spark breaks up the processing of RDD operations into tasks, each of which is executed by an executor. Prior to execution, Spark computes the task’s closure. The closure is those variables and methods which must be visible for the executor to perform its computations on the RDD (in this case foreach()). This closure is serialized and sent to each executor.

The variables within the closure sent to each executor are now copies and thus, when counter is referenced within the foreach function, it’s no longer the counter on the driver node. There is still a counter in the memory of the driver node but this is no longer visible to the executors! The executors only see the copy from the serialized closure. Thus, the final value of counter will still be zero since all operations on counter were referencing the value within the serialized closure.

11. Shipping Closures, Broadcast Variables and Accumulator Variables

Closures, Broadcast and Accumulator Variables

Spark automatically creates closures

• for functions that run on RDDs at workers,
• and for any global variables that are used by those workers
• one closure per worker is sent with every task
• and there's no communication between workers
• closures are one way from the driver to the worker
• any changes that you make to the global variables at the workers
  • are not sent to the driver or
  • are not sent to other workers.

The problem we have is that these closures

• are automatically created are sent or re-sent with every job
• with a large global variable it gets inefficient to send/resend lots of data to each worker
• we cannot communicate that back to the driver

To do this, Spark provides shared variables in two different types.

• broadcast variables
  • lets us to efficiently send large read-only values to all of the workers
  • these are saved at the workers for use in one or more Spark operations.
• accumulator variables
  • These allow us to aggregate values from workers back to the driver.
  • only the driver can access the value of the accumulator
  • for the tasks, the accumulators are basically write-only

Accumulators

Accumulators are variables that are only “added” to through an associative and commutative operation and can therefore be efficiently supported in parallel. They can be used to implement counters (as in MapReduce) or sums. Spark natively supports accumulators of numeric types, and programmers can add support for new types.

Read: https://spark.apache.org/docs/latest/rdd-programming-guide.html#accumulators.

A numeric accumulator can be created by calling SparkContext.longAccumulator() or SparkContext.doubleAccumulator() to accumulate values of type Long or Double, respectively. Tasks running on a cluster can then add to it using the add method. However, they cannot read its value. Only the driver program can read the accumulator’s value, using its value method.

The code below shows an accumulator being used to add up the elements of an array:

Broadcast Variables

From https://spark.apache.org/docs/latest/rdd-programming-guide.html#broadcast-variables:

Broadcast variables allow the programmer to keep a read-only variable cached on each machine rather than shipping a copy of it with tasks. They can be used, for example, to give every node a copy of a large input dataset in an efficient manner. Spark also attempts to distribute broadcast variables using efficient broadcast algorithms to reduce communication cost.

Spark actions are executed through a set of stages, separated by distributed “shuffle” operations. Spark automatically broadcasts the common data needed by tasks within each stage. The data broadcasted this way is cached in serialized form and deserialized before running each task. This means that explicitly creating broadcast variables is only useful when tasks across multiple stages need the same data or when caching the data in deserialized form is important.

Broadcast variables are created from a variable v by calling SparkContext.broadcast(v). The broadcast variable is a wrapper around v, and its value can be accessed by calling the value method. The code below shows this in action.

After the broadcast variable is created, it should be used instead of the value v in any functions run on the cluster so that v is not shipped to the nodes more than once. In addition, the object v should not be modified after it is broadcast in order to ensure that all nodes get the same value of the broadcast variable (e.g. if the variable is shipped to a new node later).

To release the resources that the broadcast variable copied onto executors, call .unpersist(). If the broadcast is used again afterwards, it will be re-broadcast. To permanently release all resources used by the broadcast variable, call .destroy(). The broadcast variable can’t be used after that. Note that these methods do not block by default. To block until resources are freed, specify blocking=true when calling them.

A more interesting example of broadcast variable

Let us broadcast maps and use them to lookup the values at each executor. This example is taken from: - https://sparkbyexamples.com/spark/spark-broadcast-variables/

13. HOMEWORK

See the notebook in this folder named 005_RDDsTransformationsActionsHOMEWORK. This notebook will give you more examples of the operations above as well as others we will be using later, including:

• Perform the takeOrdered action on the RDD
• Transform the RDD by distinct to make another RDD and
• Doing a bunch of transformations to our RDD and performing an action in a single cell.

14. Importing Standard Scala and Java libraries

• For other libraries that are not available by default, you can upload other libraries to the Workspace.
• Refer to the Libraries guide for more details.

To read at ease more methodically go through Chapter 12. Resilient Distributed Datasets (RDDs) of Spark: The Definitive Guide:

• https://learning.oreilly.com/library/view/spark-the-definitive/9781491912201/ch12.html

Note, you may need access via your library.

ScaDaMaLe Course site and book

HOMEWORK on RDD Transformations and Actions

Just go through the notebook and familiarize yourself with these transformations and actions. This will help you do the autograded Assignment.

1. Perform the takeOrdered action on the RDD

To illustrate take and takeOrdered actions, let's create a bigger RDD named rdd0_1000000 that is made up of a million integers from 0 to 1000000. We will sc.parallelize the Seq Scala collection by using its .range(startInteger,stopInteger) method.

takeordered(n) returns n elements ordered in ascending order (by default) or as specified by the optional key function, as shown below.

2. More examples of map

To see what's in the transformed RDD, let's perform the actions of count and collect on the rdd.map( x => x*2), the transformation of rdd by the map given by the closure x => x*2.

3. More examples of filter

Let's declare another val RDD named rddFiltered by transforming our first RDD named rdd via the filter transformation x%2==0 (of being even).

This filter transformation based on the closure x => x%2==0 will return true if the element, modulo two, equals zero. The closure is automatically passed on to the workers for evaluation (when an action is called later). So this will take our RDD of (1,2,3,4) and return RDD of (2, 4).

4. More examples of reduce

CAUTION: Know the limits of your numeric types!

The minimum and maximum value of Int and Long types are as follows:

As the following products over Long Integers indicate, they are limited too!

5. Let us do a bunch of transformations to our RDD and perform an action

• start from a Scala Seq,
• sc.parallelize the list to create an RDD,
• filter that RDD, creating a new filtered RDD,
• do a map transformation that maps that RDD to a new mapped RDD,
• and finally, perform a reduce action to sum the elements in the RDD.

This last reduce action causes the parallelize, the filter, and the map transformations to actually be executed, and return a result back to the driver machine.

6. Transform the RDD by distinct to make another RDD

Let's declare another RDD named rdd2 that has some repeated elements to apply the distinct transformation to it. That would give us a new RDD that only contains the distinct elements of the input RDD.

Let's apply the distinct transformation to rdd2 and have it return a new RDD named rdd2Distinct that contains the distinct elements of the source RDD rdd2.

7. more flatMap

Let us pass the rdd above to a map with a closure that will take in each element x and return Array(x, x+5). So each element of the mapped RDD named rddOfArrays is an Array[Int], an array of integers.

Now let's observer what happens when we use flatMap to transform the same rdd and create another RDD called rddfM.

Interestingly, flatMap flattens our rdd by taking each Array (or sequence in general) and truning it into individual elements.

Thus, we end up with the RDD rddfM consisting of the elements (1, 6, 2, 7, 3, 8) as shown from the output of rddfM.collect below.

ScaDaMaLe Course site and book

Word Count on US State of the Union (SoU) Addresses

• Word Count in big data is the equivalent of Hello World in programming
• We count the number of occurences of each word in the first and last (2016) SoU addresses.

You should have loaded data by now, i.e., run all cells in the Notebook 002_02_dbcCEdataLoader.

An interesting analysis of the textual content of the State of the Union (SoU) addresses by all US presidents was done in:

• Alix Rule, Jean-Philippe Cointet, and Peter S. Bearman, Lexical shifts, substantive changes, and continuity in State of the Union discourse, 1790–2014, PNAS 2015 112 (35) 10837-10844; doi:10.1073/pnas.1512221112.

Fig. 5. A river network captures the flow across history of US political discourse, as perceived by contemporaries. Time moves along the x axis. Clusters on semantic networks of 300 most frequent terms for each of 10 historical periods are displayed as vertical bars. Relations between clusters of adjacent periods are indexed by gray flows, whose density reflects their degree of connection. Streams that connect at any point in history may be considered to be part of the same system, indicated with a single color.

Let us investigate this dataset ourselves!

1. We first get the source text data by scraping and parsing from http://stateoftheunion.onetwothree.net/texts/index.html as explained in scraping and parsing SoU addresses.

• All of the SOU addresses from the founding of the United States until 2017 has been scraped into a single file.
• This data is already made available in our distributed file system.
• We only do the simplest word count with this data in this notebook and will do more sophisticated analyses in the sequel (including topic modeling, etc).

Key Data Management Concepts

The Structure Spectrum

Let's peruse through the following blog from IBM to get an idea of structured, unstructured and semi-structured data:

• https://www.ibm.com/cloud/blog/structured-vs-unstructured-data

In this notebook we will be working with unstructured or schema-never data (plain text files).

Later on we will be working with structured or tabular data as well as semi-structured data.

DBFS and dbutils - where is this dataset in our distributed file system inside databricks?

• Since we are on the databricks cloud, it has a file system called DBFS
• DBFS is similar to HDFS, the Hadoop distributed file system
• dbutils allows us to interact with dbfs.
• The 'display' command displays the list of files in a given directory in the file system.

Note: If you are on zeppelin or a non-databricks environment then use hdfs or drop into shell via %sh to ls the files in the local file system.

Read the file into Spark Context as an RDD of Strings

• The textFile method on the available SparkContext sc can read the gnu-zip compressed text file /datasets/sds/souall.txt.gz into Spark and create an RDD of Strings
  • but this is done lazily until an action is taken on the RDD sou!

Perform some actions on the RDD

• Each String in the RDD sou represents one line of data from the file and can be made to perform one of the following actions:
  • count the number of elements in the RDD sou (i.e., the number of lines in the text file /datasets/sds/souall.txt.gz) using sou.count
  • display the contents of the RDD using take
  • do not use collect as it will bring all the lines to the Driver, which could crash the Driver if the RDD is very large.

SInce we have several SOU addresses we can use zipWithIndex and filter to extract the specific lines in a range of interest.

Cache the RDD in (distributed) memory to avoid recreating it for each action

• Above, every time we took an action on the same RDD, the RDD was reconstructed from the textfile.
  • Spark's advantage compared to Hadoop MapReduce is the ability to cache or store the RDD in distributed memory across the nodes.
• Let's use .cache() after creating an RDD so that it is in memory after the first action (and thus avoid reconstruction for subsequent actions).
  • count the number of elements in the RDD sou (i.e., the number of lines in the text file /datasets/sds/souall.txt.gz) using sou.count()
  • display the contents of the RDD using take or collect.

Now, let's go to SparkUI and verify that this RDD is actually cached. The details would look something akin to the following.

Question

• what happens when we call the following now?

Let's quickly go back to SparkUI's Storage and see what is cached now.

What if we tried to cache souWithIndices?

Now, you will notice that both RDDs are cached in memory now.

This is merely illustrative and makes little difference in this example. However, if you do a whole lot of transformations to an RDD and plan to use it in the immediate future then it may be worthwhile caching the intermediate RDD to save time. This is often done in iterative ML algorithms.

Lifecycle of a Spark Program - Summary

• create RDDs from:
  • some external data source (such as a distributed file system)
  • parallelized collection in your driver program
• lazily transform these RDDs into new RDDs
• cache some of those RDDs for future reuse
• you perform actions to execute parallel computation to produce results

Transform lines to words

• We need to loop through each line and split the line into words
• For now, let us split using whitespace
• More sophisticated regular expressions can be used to split the line (as we will see soon)

Naive word count

At a first glace, to do a word count of George Washingtons SoU address, we are templed to do the following:

• just break each line by the whitespace character " " and find the words using a flatMap
• then do the map with the closure word => (word, 1) to initialize each word with a integer count of 1
  • ie., transform each word to a (key, value) pair or Tuple such as (word, 1)
• then count all values with the same key (word is the Key in our case) by doing a
  • reduceByKey(_+_)
• and finally collect() to display the results.

Unfortunately, as you can see from the collect above:

• the words have punctuations at the end which means that the same words are being counted as different words. Eg: importance
• empty words are being counted

So we need a bit of regex'ing or regular-expression matching (all readily available from Scala via Java String types).

We will cover the three things we want to do with a simple example from Middle Earth!

• replace all multiple whitespace characters with one white space character " "
• replace all punction characters we specify within [ and ] such as [,?.!:;] by the empty string "" (i.e., remove these punctuation characters)
• convert everything to lower-case.

More sophisticated word count

We are now ready to do a word count of George Washington's SoU on January 8th 1790 as follows:

Doing it all together for the SOU Addresses of George Washington and Donald Trump

YouTry: HOMEWORK

• HOWEWORK WordCount 1: sortBy
• HOMEWORK WordCount 2: dbutils.fs
  • for databricks environment only; won't work in zeppelin which may use hdfs or local file system or another

HOMEWORK WordCount 1. sortBy

Let's understand sortBy a bit more carefully.

Make your code easy to read for other developers ;) Use 'case classes' with well defined variable names that everyone can understand

If you just want a total count of all words in the file

YoutTry: HOMEWORK WordCount 2: dbutils.fs

Have a brief look at what other commands dbutils.fs supports. We will introduce them as needed in databricks

• This is for databricks environment only; dbutils.fs won't work in zeppelin which may use hdfs or local file system or another

Feel free to explore syntax by writing your own codes in your notebooks.

ScaDaMaLe Course site and book

Piped RDDs and More

This is a very useful but Advanced Topic.

This notebooks, demonstrates how you can go through Chapters of the recommended book Spark: The Definitive Guide and get your hands dirty with the codes in it to learn by coding, in addition to reading.

Here we will first take excerpts with minor modifications from the end of Chapter 12. Resilient Distributed Datasets (RDDs) of Spark: The Definitive Guide:

• https://learning.oreilly.com/library/view/spark-the-definitive/9781491912201/ch12.html

Next, we will do Bayesian AB Testing using PipedRDDs.

First, we create the toy RDDs as in The Definitive Guide:

From a Local Collection

To create an RDD from a collection, you will need to use the parallelize method on a SparkContext (within a SparkSession). This turns a single node collection into a parallel collection. When creating this parallel collection, you can also explicitly state the number of partitions into which you would like to distribute this array. In this case, we are creating two partitions:

glom from The Definitive Guide

glom is an interesting function that takes every partition in your dataset and converts them to arrays. This can be useful if you’re going to collect the data to the driver and want to have an array for each partition. However, this can cause serious stability issues because if you have large partitions or a large number of partitions, it’s simple to crash the driver.

Let's use glom to see how our words are distributed among the two partitions we used explicitly.

Checkpointing from The Definitive Guide

One feature not available in the DataFrame API is the concept of checkpointing. Checkpointing is the act of saving an RDD to disk so that future references to this RDD point to those intermediate partitions on disk rather than recomputing the RDD from its original source. This is similar to caching except that it’s not stored in memory, only disk. This can be helpful when performing iterative computation, similar to the use cases for caching:

Let's create a directory for checkpointing of RDDs in the sequel.

Now, when we reference this RDD, it will derive from the checkpoint instead of the source data. This can be a helpful optimization.

YouTry

Just some more words in haha_words with \n, the End-Of-Line (EOL) characters, in-place.

Let's use glom to see how our haha_words are distributed among the partitions

Pipe RDDs to System Commands

The pipe method is probably one of Spark’s more interesting methods. With pipe, you can return an RDD created by piping elements to a forked external process. The resulting RDD is computed by executing the given process once per partition. All elements of each input partition are written to a process’s stdin as lines of input separated by a newline. The resulting partition consists of the process’s stdout output, with each line of stdout resulting in one element of the output partition. A process is invoked even for empty partitions.

The print behavior can be customized by providing two functions.

We can use a simple example and pipe each partition to the command wc. Each row will be passed in as a new line, so if we perform a line count, we will get the number of lines, one per partition:

The following produces a PipedRDD:

Now, we take an action via collect to bring the results to the Driver.

NOTE: Be careful what you collect! You can always write the output to parquet of binary files in dbfs:/// if the returned output is large.

In this case, we got the number of lines returned by wc -l per partition.

YouTry

Try to make sense of the next few cells where we do NOT specifiy the number of partitions explicitly and let Spark decide on the number of partitions automatically.

Do you understand why the above collect statement returns what it does?

Did you understand why some of the results are 0 in the last collect statement?

mapPartitions

The previous command revealed that Spark operates on a per-partition basis when it comes to actually executing code. You also might have noticed earlier that the return signature of a map function on an RDD is actually MapPartitionsRDD.

Or ParallelCollectionRDD in our case.

This is because map is just a row-wise alias for mapPartitions, which makes it possible for you to map an individual partition (represented as an iterator). That’s because physically on the cluster we operate on each partition individually (and not a specific row). A simple example creates the value “1” for every partition in our data, and the sum of the following expression will count the number of partitions we have:

Naturally, this means that we operate on a per-partition basis and therefore it allows us to perform an operation on that entire partition. This is valuable for performing something on an entire subdataset of your RDD. You can gather all values of a partition class or group into one partition and then operate on that entire group using arbitrary functions and controls. An example use case of this would be that you could pipe this through some custom machine learning algorithm and train an individual model for that company’s portion of the dataset. A Facebook engineer has an interesting demonstration of their particular implementation of the pipe operator with a similar use case demonstrated at Spark Summit East 2017.

Other functions similar to mapPartitions include mapPartitionsWithIndex. With this you specify a function that accepts an index (within the partition) and an iterator that goes through all items within the partition. The partition index is the partition number in your RDD, which identifies where each record in our dataset sits (and potentially allows you to debug). You might use this to test whether your map functions are behaving correctly:

foreachPartition

Although mapPartitions needs a return value to work properly, this next function does not. foreachPartition simply iterates over all the partitions of the data. The difference is that the function has no return value. This makes it great for doing something with each partition like writing it out to a database. In fact, this is how many data source connectors are written. You can create

your

own text file source if you want by specifying outputs to the temp directory with a random ID:

You’ll find these two files if you scan your /tmp directory.

You need to scan for the file across all the nodes. As the file may not be in the Driver node's /tmp/ directory but in those of the executors that hosted the partition.