PySpark getting started

In this post I'm going to show first steps for working with next components of PySpark:  RDDs and DataFrames.

To start working with PySpark we have 2 options:
-  python spark-shell from spark distro
-  setup dev env by our own

Let's make a closer look on both of them.

Option 1: Spark-shell

Simplest way to play with pyspark is using python spark-shell. First you have to download Spark from official web page(https://spark.apache.org/downloads.html). Next, unpack it and run "pyspark" from "bin" folder. You should see something like this:

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.4
      /_/

Using Python version 2.7.15+ (default, Oct  7 2019 17:39:04)
SparkSession available as 'spark'.

And we're ready to go! Let's try some basic operations:


>>> rdd1 = spark.parallelize([('Joe',1980), ('Huan',1978), ('Max', 1985) ])

>>> rdd1.take(1)
[('Joe', 1980)]

>>> rdd1.collect()
[('Joe', 1980), ('Huan', 1978), ('Max', 1985)]

>>> rdd1.collect()[1]

('Huan', 1978)

It's definitely enough for "playing" with PySpark, but for complex applications it's not an option: we need to setup dev env.

Option 2: setup PySpark dev env

Python setup

In my ubuntu I have both pythons 2 and 3 installed, but default is 2. To use python 3 I updated bash profile file:   ~/.bashrc : I added line
alias python=python3

Let's check:
~$ python --version
Python 3.6.8

Great! 

Pip3 

Next, let's install pip3: 

sudo apt-get -y install python3-pip

For me, after installation pip3 was not working, so I had to modify file /usr/bin/pip3 :

#from pip import main

from pip import __main__

if __name__ == '__main__':

#    sys.exit(main())

    sys.exit(__main__._main())

And let's check it: 

~$ pip3 --version
pip 19.2.3 from /home/dmitry/.local/lib/python3.6/site-packages/pip (python 3.6)

It works! 

Pipenv

Now let's install pipenv using just installed pip3:
~$ sudo pip3 install pipenv
Installing collected packages: pipenv
Successfully installed pipenv-2018.11.26

Dev env

And now we're finally ready to go! 

Next let's create a folder for our test application:

mkdir pyspark-test

cd pyspark-test

Now let's create a virtual env using pipenv: 

pyspark-test$ pipenv shell

Creating a virtualenv for this project…

✔ Successfully created virtual environment! 

Next, we can try to add some dependencies:
(pyspark-test) pyspark-test$ pipenv install pyspark
Installing pyspark…
✔ Success! 
Updated Pipfile.lock (1869ad)!
Installing dependencies from Pipfile.lock (1869ad)…

Let's check:
(pyspark-test) pyspark-test$ pip3 freeze
py4j==0.10.7
pyspark==2.4.4

- looks good: all dependencies are in place.

Finally everything is prepared for coding! I created folder "src" and file "main.py" inside with next content:

from pyspark.sql import SparkSession

if __name__ == '__main__':

        spark = SparkSession.builder.appName("MyTestApp").getOrCreate()

        rdd1 = spark.sparkContext.parallelize([('Joe',1980), ('Huan',1978), ('Max', 1985) ])

        print("Count: ", rdd1.count())

        print("First: ", rdd1.take(1))

Let's run it now:

(pyspark-test) pyspark-test$ python ./src/main.py 

Count:  3                                                                       

First:  [('Joe', 1980)]

Cool! Our environment is ready for coding! 

RDD

First it's better to read about RDD from official web page: https://spark.apache.org/docs/latest/rdd-programming-guide.html

In shorts:
The main abstraction Spark provides is a resilient distributed dataset (RDD), which is a collection of elements partitioned across the nodes of the cluster that can be operated on in parallel. RDDs are created by starting with a file in the Hadoop file system (or any other Hadoop-supported file system), or an existing Scala collection in the driver program, and transforming it. Users may also ask Spark to persist an RDD in memory, allowing it to be reused efficiently across parallel operations. Finally, RDDs automatically recover from node failures.



In a code provided below I'm showing operations which can performed on RDDs:
- rdd creation: in memory and reading from file
- rdd actions: count, take
- rdd transformations: map, flatMap, filter

from pyspark.sql import SparkSession

def main():

    # init spark

    spark = SparkSession.builder.appName("MyTestApp").getOrCreate()

    sc = spark.sparkContext

    # create rdd "in-memory"

    rdd1 = sc.parallelize([('Joe', 1980), ('Huan', 1978), ('Max', 1985)])

    print("RDD1 Count: ", rdd1.count()) #RDD1 Count:  3 

    print("RDD1 First: ", rdd1.take(1)) #RDD1 First:  [('Joe', 1980)]

    # create rdd by reading from file

    rdd2 = sc.textFile("test/test-data/groups.csv") 

    print("RDD2 All: ", rdd2.take(10)) #RDD2 All:  ['101,Admin', '102,Dev', '103,DB']

    # RDD transformation using MAP method

    rdd2formatted = rdd2 \

        .map(lambda line: line.split(",")) \

        .map(lambda arr: (arr[0], arr[1]))

    print("RDD2Formatted All: ", rdd2formatted.take(10)) 

    #RDD2Formatted All:  [('101', 'Admin'), ('102', 'Dev'), ('103', 'DB')]

    # RDD filtering using FILTER method

    rdd2filtered = rdd2formatted.filter(lambda row: int(row[0]) > 101)

    print("RDD2Filtered All: ", rdd2filtered.take(10))

    #RDD2Filtered All:  [('102', 'Dev'), ('103', 'DB')]

    # Flattening RDD

    rdd2FilteredFlat = rdd2filtered.flatMap(lambda row: (row[0], row[1]))

    print("RDD2FilteredFlat All: ", rdd2FilteredFlat.take(10))

    #RDD2FilteredFlat All:  ['102', 'Dev', '103', 'DB']

if __name__ == '__main__':

    main()

p.s. content of "test/test-data/groups.csv" is following:
101,Admin
102,Dev
103,DB

DataFrame

First, I would again suggest to read official documentation: https://spark.apache.org/docs/latest/sql-programming-guide.html.
In shorts, if you know what is RDD -  it's very easy to understand what is DataFrame: it'a a RDD + Schema. Where schema is an information about field names and types. If we have structured data like JSON, CSV - we can just read them using spark and it will take the schema from files(in case of CSV - from header).

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. 


Below I'm showing: how to create dataFrame, some basic operations, how to join them. Most interesting thing here: there are to ways for data transformtion here:
- using Spark API
- using Spark SQL

from pyspark.sql import SparkSession

def main():

    # init spark

    spark = SparkSession.builder.appName("MyTestApp").getOrCreate()

    sc = spark.sparkContext

    # create DataFrame from RDD

    rdd1 = sc.parallelize((

    """{"id": "1001", "name": "Joe",  "depId": 101}""",

    """{"id": "1002", "name": "Huan", "depId": 102}""", 

    """{"id": "1003", "name": "Max",  "depId": 103}"""

    ))

    print(rdd1.take(2)) 

    # ['{"id": "1001", "name": "Joe",  "depId": 101}', '{"id": "1002", "name": "Huan", "depId": 102}']    

    df1 = spark.read.json(rdd1)

    df1.show()

    # +-----+----+----+

    # |depId|  id|name|

    # +-----+----+----+

    # |  101|1001| Joe|

    # |  102|1002|Huan|

    # |  103|1003| Max|

    # +-----+----+----+

    # Basic operations using Spark API

    df1.select("name", "depId").where("id='1001'").show()

    #+----+-----+

    #|name|depId|

    #+----+-----+

    #| Joe|  101|

    #+----+-----+    

    # Basic operation using Spark SQL

    df1.createOrReplaceTempView("users")

    spark.sql("SELECT concat(name,' has id>1001') as user_name FROM users WHERE id>1001").show()

    #+----------------+

    #|       user_name|

    #+----------------+

    #|Huan has id>1001|

    #| Max has id>1001|

    #+----------------+

    # create DataFrame by reading from file

    df2 = spark.read.option("header",True).csv("test/test-data/depts.csv")

    df2.show()

    #+-----+-------+

    #|depId|depName|

    #+-----+-------+

    #|  101|  Admin|

    #|  102|    Dev|

    #|  103|     DB|

    #+-----+-------+

    # join 2 DataFrames using Spark API

    joined1 = df1.join(df2, df1.depId==df2.depId).drop(df2.depId)

    joined1.show()

    #+-----+----+----+-------+

    #|depId|  id|name|depName|

    #+-----+----+----+-------+

    #|  101|1001| Joe|  Admin|

    #|  102|1002|Huan|    Dev|

    #|  103|1003| Max|     DB|

    #+-----+----+----+-------+    

    # join 2 DataFrames using Spark SQL

    df2.createOrReplaceTempView("deps")

    joined2 = spark.sql("SELECT u.id, u.name, d.depName FROM users u, deps d WHERE u.depId = d.depId")

    joined2.show()

    #+----+----+-------+

    #|  id|name|depName|

    #+----+----+-------+

    #|1001| Joe|  Admin|

    #|1002|Huan|    Dev|

    #|1003| Max|     DB|

    #+----+----+-------+

if __name__ == '__main__':

    main()

The end

And that is basically it. Of course it's impossible to show all RDD and DataFrame stuff in one post: it would be a book - so I just tried to show some very basic stuff for understanding "who is who" :)

Apache Spark - getting started: batch and stream data processing using scala

0. Intro 

In this post I'm going to explain basics of Apache Spark: RDD, SparkSQL, DataFrames, SparkStreaming. Actually Spark -  it's just a library for data processing. So it can be executed without any BigData-related stuff. You can just run code from this post without any Hadoop/HDFS.

1. What is Apache Spark

from wiki:
Apache Spark is an open-source cluster-computing framework. Originally developed at the University of California, Berkeley's AMPLab, the Spark codebase was later donated to the Apache Software Foundation, which has maintained it since. Spark provides an interface for programming entire clusters with implicit data parallelism and fault tolerance.

From official documentaion:
Apache Spark is a fast and general-purpose cluster computing system. It provides high-level APIs in Java, Scala, Python and R, and an optimized engine that supports general execution graphs. It also supports a rich set of higher-level tools including Spark SQL for SQL and structured data processing, MLlib for machine learning, GraphX for graph processing, and Spark Streaming.

As you can see, there is no mentioning of Hadoop/HDFS at all. It CAN work with Hadoop/HDFS, it CAN work with cluster resource manager like YARN. But for getting familiar,  also it can be used  for processing local files in standalone mode.

2. What is Lambda architecture

From wiki:
Lambda architecture is a data-processing architecture designed to handle massive quantities of data by taking advantage of both batch- and stream-processingmethods. This approach to architecture attempts to balance latency, throughput, and fault-tolerance by using batch processing to provide comprehensive and accurate views of batch data, while simultaneously using real-time stream processing to provide views of online data. The two view outputs may be joined before presentation. The rise of lambda architecture is correlated with the growth of big data, real-time analytics, and the drive to mitigate the latencies of map-reduce.^[1]

In shorts, lambda architecture is a combination of 2 processing types:
- slow but precese - in this post we will use Spark RDD/DataFrame for it
- fast but not precise -  in this post we will use Spark Streaming for it

3. Project setup: pom.xml

Here I'm using maven with scala. In pom.xml I'm adding dependencies for spark-core, spark-sql, spark-streaming.

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.demien</groupId>
    <artifactId>sparktest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>


    </dependencies>


</project>

4. Project structure

As you can see, we will be using 2 processing types of batch processing: RDD and DataFrame. And one type of stream processing: DStream. Apart from that we will be using SparkUtils class for creation of common sprak entities: SparkContext, SparkSession. And a generator of text files(for streaming) TextFileCreator.

5. SparkUtils

As I mentioned before, we will be using SparkUtils class for creation of common sprak entities: SparkContext, SparkSession:

package com.demien.sparktest

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

object SparkUtils {

  val DEF_APP_NAME = "MySparkApp"  val DEF_MASTER = "local[2]"
  //The SparkContext object - connection to a Spark execution environment and created RDDs  def getSparkContext(appName: String, master: String): SparkContext = new SparkContext(new SparkConf().setAppName(appName).setMaster(master))

  def getSparkContext(): SparkContext = getSparkContext(DEF_APP_NAME, DEF_MASTER)

  //The SparkSession - connection to dataframes and SQLs  def getSparkSession(appName: String, master: String): SparkSession = SparkSession
    .builder()
    .appName(appName)
    .master(master)
    .getOrCreate()

  def getSparkSession(): SparkSession = getSparkSession(DEF_APP_NAME, DEF_MASTER)


}

6. Batch processing: RDD

From official documentation:

The main abstraction Spark provides is a resilient distributed dataset (RDD), which is a collection of elements partitioned across the nodes of the cluster that can be operated on in parallel. RDDs are created by starting with a file in the Hadoop file system (or any other Hadoop-supported file system), or an existing Scala collection in the driver program, and transforming it. Users may also ask Spark to persist an RDD in memory, allowing it to be reused efficiently across parallel operations. Finally, RDDs automatically recover from node failures.

RDD Example: 

package com.demien.sparktest.batch

import com.demien.sparktest.SparkUtils
import org.apache.spark.rdd.RDDwith

// https://spark.apache.org/docs/2.3.0/rdd-programming-guide.htmlobject RddExample extends App {

  val sc = SparkUtils.getSparkContext()
  val file = sc.textFile("src/main/resources/sample.txt")
  val words: RDD[String] = file.flatMap(l => l.split(" ")).filter(w => w.length > 1)
  val pairs: RDD[(String, Int)] = words.map(s => (s, 1)) // [the, of, the] => (the, 1) (of, 1) (the, 1)  val counts: RDD[(String, Int)] = pairs.reduceByKey((a, b) => a + b) // (the, 1) (of, 1) (the, 1) => (the, 2) (of, 1)  val countByWord: RDD[(Int, String)] = counts.map(p => p.swap) // (the, 2) (of, 1) => (2, the) (1, of)  val countByWordSorted: RDD[(Int, String)] = countByWord.sortByKey(false)
  val top5 = countByWordSorted.take(5)

  top5.foreach(p => println(p))
}

- I added comments and  datatypes for RDD variables(which of cource are not needed here) to make it more clear. This RDD example is processing sample text file - it's just a text from ApacheSpark wiki. We are splitting text into words, creating for every work "paired object" with word itself and number 1. After that, we are groupping these pairs using word as a key and counting provided numbers.

Results:
(55,the)
(46,of)
(43,Spark)
(39,and)
(24,in)

- as you can see, most popular word(excluding "the", "of", "and", "in") in Spark wiki is "Spark" :)

7. Batch processing: DataFrame/Spark SQL

From official documentation:

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs. The DataFrame API is available in Scala, Java, Python, and R. In Scala and Java, a DataFrame is represented by a Dataset of Rows. In the Scala API, DataFrame is simply a type alias of Dataset[Row]. While, in Java API, users need to use Dataset<Row> to represent a DataFrame.

DataFrames - are the structured datasets, so as a sample file we will be using not TEXT file but  JSON like this:

{
  "name": "Keeley Bosco",
  "email": "katlyn@jenkinsmaggio.net",
  "city": "Lake Gladysberg",
  "mac": "08:fd:0b:cd:77:f7",
  "timestamp": "2015-04-25 13:57:36 +0700",
  "creditcard": "1228-1221-1221-1431"}
{
  "name": "Rubye Jerde",
  "email": "juvenal@johnston.name",
  "city": null,
  "mac": "90:4d:fa:42:63:a2",
  "timestamp": "2015-04-25 09:02:04 +0700",
  "creditcard": "1228-1221-1221-1431"}

DataFrame example:

package com.demien.sparktest.batch

// https://spark.apache.org/docs/latest/sql-programming-guide.html
import org.apache.spark.sql.SparkSession

object DataFrameExample extends App {

  val spark = SparkSession
    .builder()
    .appName("Spark SQL basic example")
    .config("spark.master", "local")
    .getOrCreate()

  // For implicit conversions like converting RDDs to DataFrames  import spark.implicits._


  val df = spark.read.json("src/main/resources/people.json")
  df.printSchema()
  df.createOrReplaceTempView("people")
  val sqlDF = spark.sql("SELECT * FROM people where email like '%net%' ")
  sqlDF.show()

  case class Person(name: String, email: String, city: String, mac: String, timestamp: String, creditcard: String)

  val peopleDS = spark.read.json("src/main/resources/people.json").as[Person]
  val filteredDS = peopleDS.filter(p => p.email != null && p.email.contains("net"))
  filteredDS.show()


}

We are using SparkSQL to query our structured dataset(DataFrame) for people which have "%net%" in their emails. Also, at the end we are doing the same thing again, but using using DataSet api.
Or cource, in both cases results are the same:

+---------------+-------------------+--------------------+-----------------+----------------+--------------------+
|           city|         creditcard|               email|              mac|            name|           timestamp|
+---------------+-------------------+--------------------+-----------------+----------------+--------------------+
|Lake Gladysberg|1228-1221-1221-1431|katlyn@jenkinsmag...|08:fd:0b:cd:77:f7|    Keeley Bosco|2015-04-25 13:57:...|
|           null|1228-1221-1221-1431|emery_kunze@rogah...|3a:af:c9:0b:5c:08|Celine Ankunding|2015-04-25 14:22:...|
+---------------+-------------------+--------------------+-----------------+----------------+--------------------+


Unfortunatelly, spark is not showing full values, but these email values  are:

"katlyn@jenkinsmaggio.net, "emery_kunze@rogahn.net"

- emails which contain "net".

8. Stream processing: DStream

From official documentation:

Spark Streaming is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams. Data can be ingested from many sources like Kafka, Flume, Kinesis, or TCP sockets, and can be processed using complex algorithms expressed with high-level functions like map, reduce, join and window. Finally, processed data can be pushed out to filesystems, databases, and live dashboards. In fact, you can apply Spark’s machine learning and graph processing algorithms on data streams.

8.1 Stream processing: DStream: TextFileCreator

To simulate stream of data, we will create the simple application which is creating text file every 10 seconds. As a source for this file I will be using again text from ApachSpark wiki. 

package com.demien.sparktest

import java.io.FileWriter
import java.util.Date

import scala.io.Source
import scala.util.Random

object TextFileCreator extends App {

  val listOfLines = Source.fromFile("src/main/resources/sample.txt").getLines.toList
  val rnd = new Random()

  while (true) {

    val fileName = new Date().getTime
    val fullFileName = "data/" + fileName + ".txt"    val fw = new FileWriter(fullFileName, true)
    println("writing to " + fullFileName)

    val linesCount = rnd.nextInt(20) + 5    for (i <- 1 to linesCount) fw.write(listOfLines(rnd.nextInt(100)) + "\n")

    fw.close()
    Thread.sleep(10000)

  }

}

8.2 Stream processing: DStream: Streaming example itself

Our application will be monitoring "data" folder for new files. When new file is received - it will be processed. To simulate some statefull activity we will be using function:  specFunc - the point is to constantly calculate count of words. 

package com.demien.sparktest.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}

object DStreamExample extends App {


  val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  val ssc = new StreamingContext(conf, Seconds(10))
  ssc.checkpoint("spark-checkpoint")

  val lines = ssc.textFileStream("data")
  val words = lines.flatMap(_.split(" "))
  val pairs = words.map(word => (word, 1))
  val wordCounts = pairs.reduceByKey(_ + _)


  def specFunc = (key: String, value: Option[Int], state: State[Int]) => {
    var newState = state.getOption().getOrElse(0)
    var newValue = value.getOrElse(1)
    newState = newState + newValue
    state.update(newState)
    (key, newValue)
  }

  val spec = StateSpec.function(specFunc).timeout(Seconds(30))

  val wordsMapped = wordCounts.mapWithState(spec)

  // top 10  wordsMapped.stateSnapshots().foreachRDD(rdd => {
    rdd.map(e => (e._2, e._1)).sortByKey(false).take(10).foreach(e => print(e._1, e._2))

  })

  ssc.start() // Start the computation  ssc.awaitTermination() // Wait for the computation to terminate
}

8.3. Stream processing: DStream: Execution

Of course, we have to run both: TextFileCreator and DStreamExample.

TextFileCreator is creating files:
writing to data/1524921644177.txt
writing to data/1524921654262.txt
writing to data/1524921664264.txt
writing to data/1524921674266.txt
writing to data/1524921684267.txt
writing to data/1524921694268.txt
writing to data/1524921704270.txt


And DStreamExample is processing them and counting words:

.....
(17,a)(16,Spark)(15,)(14,the)(10,Apache)(10,of)(7,//)(5,is)(5,can)(5,in)
(17,)(17,a)(17,Spark)(14,the)(10,Apache)(10,of)(7,//)(5,is)(5,can)(5,in)
(25,)(24,the)(21,Spark)(21,a)(19,of)(12,Apache)(10,can)(10,as)(10,and)(9,is)
(35,)(27,the)(25,Spark)(25,a)(20,of)(17,and)(14,Apache)(12,as)(12,//)(10,can)
(45,)(27,the)(26,Spark)(25,a)(20,of)(17,and)(14,Apache)(12,as)(12,//)(10,can)

As you can see, results are similar to what we had in RddExample: most popular words are Spark and Apache.

9. The end. 

As you can see, to try Apache Spark you don't need Hadoop/Yarn - it's possible to run it in a standalone mode without all these compicated things. Source code can be downloaded from here.