OOP in Python

In this post I'm going to show some basics of OOP in Python.

1. Classes (which are actually objects) and access modifiers

To define class in Python we should unsurprisingly use keyword class:

class User:

To use constructor we should use "init" function:

def __init__(self, name, age):

The first parameter: self is a reference to object instance. Next parameters are actually constructor parameters. Thus we have 2 parameters here and we should define them in instance variable:

self.name = name

self.age = age

But what about visibility of them? public? private? Actually everything is public. We can use some "hacks" for private:
 - we can add leading underscore - it's just a convention that field is private, nothing is technically stopping us for accessing and changing this field
 - we can add 2 leading underscores - this is better option: Python will rename this field by pattern "_className__paramName". So field will not be visible by it's name, but will still be visible by pattern i just described.

Let's try:

class User:

def __init__(self, name, age):

self.name = name # public field

self._name = name # naive private field

self.__name = name # better private field

self.__age = age # field defined as "property" with custom SETTER

def sayHi(self):

print("Hello, I'm {} and I'm {} years old.".format(self.__name, self.__age))

And now let's play with it. Outputs are shown as comments:

joe = User("Joe", 25)

joe.sayHi() # Hello, I'm Joe and I'm 25 years old.

# playing with public field

print(joe.name) # Joe

joe.name = "Joe1"

print(joe.name) # Joe1

# playing with naive private field

print(joe._name) # Joe

joe._name = "Joe2"

print(joe._name) # Joe2

# playing with better private field

#print(joe.__name) # AttributeError: 'User' object has no attribute '__name'

print(joe._User__name)

joe._User__name = "Joe3"

print(joe._User__name) # Joe3

As you can see, everything is actually public but using 2 underscores we have at least some protection.

2. Addons provided by decorators

More advanced things are provided by decorators. For example if we need something like static method which of course don't have instance reference (self), we can do it this way:

@staticmethod

def say(msg):

print("I'm saying:", msg)

Let's try it:

User.say("Yo!") # I'm saying: Yo!

Also we can use additional decorators for having something like getter and setter:

@property

def age(self):

return self.__age

@age.setter

def age(self, newValue):

print("age can not be changed!")

That is more interesting from a "modifiers" point of view: we can not make variable invisible, but at least can protect them from being changed()!

Let's see:

print(joe.age) # 25

joe.age = 55 # age can not be changed!

print(joe.age) # 25

joe._User__age = 66

print(joe.age) # 66

When we tried to use field directly: protection worked fine. But when we tried to access used pattern ..... protection failed: we successfully changed the value.

3. Inheritance

To extend a class we should provide it name in bracers:

class SuperUser(User):

to call some method from patent classes we can use method "super()":

class SuperUser(User):

def __init__(self, name, age, role):

super().__init__(name, age)

self.__role = role

def sayHi(self):

super().sayHi()

print(" and my role is: {}".format(self.__role))

Let's try it:

admin = SuperUser("Sysdba", 44, "IT")

admin.sayHi() # Hello, I'm Sysdba and I'm 44 years old.

# and my role is: IT

With inheritance everything is more or less as expected.

4. The end

Of course, Python is not a fully OOP language but we still use some basics OOP stuff.

PySpark data preparation

In previous post( PySpark getting started) I was explaining how to setup PySpark dev env and some basic operations. This post is some kind of next step: then everything is in place: we have some data and we are about to start some interesting stuff like processing/modeling, but first, we have to prepare our data for this.

Starting point of my simple program

To start, we should import some dependencies from PySpark library:

from pyspark.sql import SparkSession
import pyspark.sql.functions as sqlfun

if __name__ == "__main__":
    spark = SparkSession.builder.appName("MyDataPrepareApp").getOrCreate()
    spark.sparkContext.setLogLevel('WARN')

Test data 

Let's generate some simple dataset:

print("Initial dataset:")
df = spark.createDataFrame([
    (1, "Joe", 34, None),
    (1, "Joe", 34, None),
    (2, "Huan", 36, 18000),
    (3, "Huan", 36, 18000),
    (4, None, 38, 19000),
    (5, "Sebastyan", 35, None),
    (7, "Anna", 30, 14000),
    (8, None, 24, 11000),
    (9, "Jordan", None, 15000),
    (10, None, 28, 13000),

    (1001, "ExtraLow", 1, 1),
    (1002, "ExtraBig", 80, 50000),
], schema=["id", "name", "age", "salary"])
df.show()

Output: 

Initial dataset:
+----+---------+----+------+
|  id|     name| age|salary|
+----+---------+----+------+
|   1|      Joe|  34|  null|
|   1|      Joe|  34|  null|
|   2|     Huan|  36| 18000|
|   3|     Huan|  36| 18000|
|   4|     null|  38| 19000|
|   5|Sebastyan|  35|  null|
|   7|     Anna|  30| 14000|
|   8|     null|  24| 11000|
|   9|   Jordan|null| 15000|
|  10|     null|  28| 13000|
|1001| ExtraLow|   1|     1|
|1002| ExtraBig|  80| 50000|
+----+---------+----+------+

Duplicates 

As you can see there are some(2 records with ID=1) duplicates in our dataset. Let's try to find and remove them:

# drop duplicatesdfDistinct = df.dropDuplicates()
print("Distinct:")
dfDistinct.show()

Output: 

Distinct:
+----+---------+----+------+
|  id|     name| age|salary|
+----+---------+----+------+
|1001| ExtraLow|   1|     1|
|   9|   Jordan|null| 15000|
|   4|     null|  38| 19000|
|   1|      Joe|  34|  null|
|   3|     Huan|  36| 18000|
|   8|     null|  24| 11000|
|   2|     Huan|  36| 18000|
|   7|     Anna|  30| 14000|
|1002| ExtraBig|  80| 50000|
|  10|     null|  28| 13000|
|   5|Sebastyan|  35|  null|
+----+---------+----+------+

It's better now: we don't have duplicates with ID=1 - just one record left. But we also have 2 rows which have different IDs (2 and 3) but in fact they are the same. Let's try to find duplicates for columns other than ID:

# drop duplicates other than ID columncolsExceptID = [col for col in df.columns if col != "id"]
print("Columns except [id]:", colsExceptID, ". Making distinct by them:")
dfDistinctExId = dfDistinct.dropDuplicates(colsExceptID)
dfDistinctExId.show()

Output:
Columns except [id]: ['name', 'age', 'salary'] . Making distinct by them:
+----+---------+----+------+
|  id|     name| age|salary|
+----+---------+----+------+
|   3|     Huan|  36| 18000|
|   1|      Joe|  34|  null|
|  10|     null|  28| 13000|
|   9|   Jordan|null| 15000|
|1001| ExtraLow|   1|     1|
|   7|     Anna|  30| 14000|
|   4|     null|  38| 19000|
|   8|     null|  24| 11000|
|   5|Sebastyan|  35|  null|
|1002| ExtraBig|  80| 50000|
+----+---------+----+------+

Good: just records with ID=3 remained (with ID=2 was removed)

Some basic aggregation

If we need to calculate count of duplicates and distinct records we can use aggregation functions:

# some aggregationsprint("Some basic aggregations:")
df.agg(
    sqlfun.count("id").alias("total_count"),
    sqlfun.countDistinct("id").alias("distinct_id"),
    sqlfun.countDistinct("name").alias("distinct_name")
).show()

Output: 

Some basic aggregations:
+-----------+-----------+-------------+
|total_count|distinct_id|distinct_name|
+-----------+-----------+-------------+
|         12|         11|            7|
+-----------+-----------+-------------+

Missing values

As we can see, some values in dataset are missing. Let's calculate how much records are missed for example  in AGE column:

# dealing with missed valuesprint("Percentage of missing values for [age] column")
dfDistinctExId.agg(
    (100 - sqlfun.count("age") / sqlfun.count("*") * 100).alias("age_missing")
).show()

Output:

+-----------+
|age_missing|
+-----------+
|       10.0|
+-----------+

- 10% are missing in AGE column.


We can also use more generic way to calculate missing values in several columns in "one shot":

print("Percentage of missing values for all columns")
dfDistinctExId.agg(*[  # symbol * here instruct spark to use each element of array as separate value    (100 - sqlfun.count(col) / sqlfun.count("*") * 100).alias(col + "_missing")
    for col in dfDistinctExId.columns
]).show()

Output: 

Percentage of missing values for all columns
+----------+------------+-----------+--------------+
|id_missing|name_missing|age_missing|salary_missing|
+----------+------------+-----------+--------------+
|       0.0|        30.0|       10.0|          20.0|
+----------+------------+-----------+--------------+

Filling empty values

Now when we identified missing values, we can try to fill them. Most popular approach here is to fill them by column median (mean). Let's calculate median(mean) for every column:

# calculate meanmeans = dfDistinctExId.agg(*[
    sqlfun.mean(col).alias(col + "_mean")
    for col in dfDistinctExId.columns if col != "id" and col != "name"])
print("Calculated medians(means):")
means.show()

Output: 

Calculated medians(means):
+--------+-----------+
|age_mean|salary_mean|
+--------+-----------+
|    34.0|  17500.125|
+--------+-----------+

Actually this form is good for us, but for PySpark it's better to transform it into "dictionary" format:

# creating dictionary for means

meansDict = {}
for i in range(len(means.columns)):
    colName = means.columns[i]
    colNameForDict = colName.replace("_mean", "")
    meanValue = means.first()[i]
    meansDict[colNameForDict] = meanValue
meansDict["name"] = "Unknown"print("Dictionary which will be used to fill missing values:")
print(meansDict)

Output:

Dictionary which will be used to fill missing values:
{'age': 34.0, 'salary': 17500.125, 'name': 'Unknown'}



For "name" column we can't use median value so we will be using word "Unknown".

And finally, using dictionary from above we can fill empty values:

dfFull = dfDistinctExId.fillna(meansDict)
print("With filled missing values:")
dfFull.show()

Output:

With filled missing values:
+----+---------+---+------+
|  id|     name|age|salary|
+----+---------+---+------+
|   3|     Huan| 36| 18000|
|   1|      Joe| 34| 17500|
|  10|  Unknown| 28| 13000|
|   9|   Jordan| 34| 15000|
|1001| ExtraLow|  1|     1|
|   7|     Anna| 30| 14000|
|   4|  Unknown| 38| 19000|
|   8|  Unknown| 24| 11000|
|   5|Sebastyan| 35| 17500|
|1002| ExtraBig| 80| 50000|
+----+---------+---+------+

Outliers

Great ! We don't have any empty values! But what about the other values? Some values (with IDs 1001 and 1002) looks too big and we also have some which are too small: outliers. Let's found and remove them. We're interesting here at columns AGE and SALARY. First we have to calculate quantiles: 0.5 - mean median (mean) of dataset , 0.25 - median of subset from the beginning to median, 0,75 - median of subset from median to the end..... In code I'm calculating them all (0.0, 0.25, 0.5, 0.75, 1.0) ,  bu actually we need just 2 of them: Q1(0.25) and Q3(0.75). Later we have to calculate IQR as difference between them and using IQR we can calculate lowest and highest threshold boundaries. Values are lower than lowest threshold or higher than higher threshold should be removed.

# calculate threshold boundaries for several columns(age and salary) and filter datacols = ["age", "salary"]

dfFiltered = dfFull
for currentCol in cols:
    qs = dfFull.approxQuantile(currentCol, [0.0, 0.25, 0.5, 0.75, 1.0], 0.05)
    Q1 = qs[1]
    Q3 = qs[3]
    IRQ = Q3 - Q1
    thresholdLow = Q1 - 1.5 * IRQ
    thresholdHigh = Q3 + 1.5 * IRQ
    print("For", currentCol, "low threshold is", thresholdLow, " and high threshold is", thresholdHigh)
    dfFiltered = dfFiltered.filter(
        (sqlfun.col(currentCol) > thresholdLow) & (sqlfun.col(currentCol) < thresholdHigh))

print("Filtered by thresholds:")
dfFiltered.show()

Output: 

For age low threshold is 16.0  and high threshold is 48.0
For salary low threshold is 5500.0  and high threshold is 25500.0
Filtered by thresholds:
+---+---------+---+------+
| id|     name|age|salary|
+---+---------+---+------+
|  3|     Huan| 36| 18000|
|  1|      Joe| 34| 17500|
| 10|  Unknown| 28| 13000|
|  9|   Jordan| 34| 15000|
|  7|     Anna| 30| 14000|
|  4|  Unknown| 38| 19000|
|  8|  Unknown| 24| 11000|
|  5|Sebastyan| 35| 17500|
+---+---------+---+------+


As we wanted values where had extra big/low values were removed.

The end

We removed duplicates and outliers, filled empty values - our data are ready for future processing / modeling ! :)

Just in case full code:

from pyspark.sql import SparkSession
import pyspark.sql.functions as sqlfun

if __name__ == "__main__":
    spark = SparkSession.builder.appName("MyDataPrepareApp").getOrCreate()
    spark.sparkContext.setLogLevel('WARN')

    print("Initial dataset:")
    df = spark.createDataFrame([
        (1, "Joe", 34, None),
        (1, "Joe", 34, None),
        (2, "Huan", 36, 18000),
        (3, "Huan", 36, 18000),
        (4, None, 38, 19000),
        (5, "Sebastyan", 35, None),
        (7, "Anna", 30, 14000),
        (8, None, 24, 11000),
        (9, "Jordan", None, 15000),
        (10, None, 28, 13000),

        (1001, "ExtraLow", 1, 1),
        (1002, "ExtraBig", 80, 50000),
    ], schema=["id", "name", "age", "salary"])
    df.show()


    # drop duplicates    dfDistinct = df.dropDuplicates()
    print("Distinct:")
    dfDistinct.show()

    # drop duplicates other than ID column    colsExceptID = [col for col in df.columns if col != "id"]
    print("Columns except [id]:", colsExceptID, ". Making distinct by them:")
    dfDistinctExId = dfDistinct.dropDuplicates(colsExceptID)
    dfDistinctExId.show()

    # some aggregations    print("Some basic aggregations:")
    df.agg(
        sqlfun.count("id").alias("total_count"),
        sqlfun.countDistinct("id").alias("distinct_id"),
        sqlfun.countDistinct("name").alias("distinct_name")
    ).show()

    # add new id column    print("With new created ID column")
    dfNewId = dfDistinctExId.withColumn("new_id", sqlfun.monotonically_increasing_id())
    dfNewId.show()

    # dealing with missed values    print("Percentage of missing values for [age] column")
    dfDistinctExId.agg(
        (100 - sqlfun.count("age") / sqlfun.count("*") * 100).alias("age_missing")
    ).show()

    print("Percentage of missing values for all columns")
    dfDistinctExId.agg(*[  # symbol * here instruct spark to use each element of array as separate value        (100 - sqlfun.count(col) / sqlfun.count("*") * 100).alias(col + "_missing")
        for col in dfDistinctExId.columns
    ]).show()

    ####################################    # filling missing values with MEAN    ####################################
    # calculate mean    means = dfDistinctExId.agg(*[
        sqlfun.mean(col).alias(col + "_mean")
        for col in dfDistinctExId.columns if col != "id" and col != "name"    ])
    print("Calculated medians(means):")
    means.show()

    # creating dictionary for means    meansDict = {}
    for i in range(len(means.columns)):
        colName = means.columns[i]
        colNameForDict = colName.replace("_mean", "")
        meanValue = means.first()[i]
        meansDict[colNameForDict] = meanValue
    meansDict["name"] = "Unknown"    print("Dictionary which will be used to fill missing values:")
    print(meansDict)

    # filling empty values using means from dictionary    dfFull = dfDistinctExId.fillna(meansDict)
    print("With filled missing values:")
    dfFull.show()

    ###########    # outliers    ###########
    # calculate quartiles: Q0, Q1, Q2, Q3, Q4 for [age] column    

    qs = dfFull.approxQuantile("age", [0.0, 0.25, 0.5, 0.75, 1.0], 0.05)  

    # last parameter - "accurance" it can be 0 but it will be really expensive to calculate it    

    print("quartiles: Q0, Q1, Q2, Q3, Q4 for [age] column:", qs)

    # calculate threshold boundaries for several columns(age and salary) and filter data    

    cols = ["age", "salary"]

    dfFiltered = dfFull
    for currentCol in cols:
        qs = dfFull.approxQuantile(currentCol, [0.0, 0.25, 0.5, 0.75, 1.0], 0.05)
        Q1 = qs[1]
        Q3 = qs[3]
        IRQ = Q3 - Q1
        thresholdLow = Q1 - 1.5 * IRQ
        thresholdHigh = Q3 + 1.5 * IRQ
        print("For", currentCol, "low threshold is", thresholdLow, " and high threshold is", thresholdHigh)
        dfFiltered = dfFiltered.filter(
            (sqlfun.col(currentCol) > thresholdLow) & (sqlfun.col(currentCol) < thresholdHigh))

    print("Filtered by thresholds:")

    dfFiltered.show()

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" :)

Python: getting started

Intro

Last time Python is getting more and more popular in BigData word.
From wiki:

Python is an interpreted high-level programming language for general-purpose programming. Created by Guido van Rossum and first released in 1991, Python has a design philosophy that emphasizes code readability, and a syntax that allows programmers to express concepts in fewer lines of code,^[26][27] notably using significant whitespace. It provides constructs that enable clear programming on both small and large scales.^[28]

Python features a dynamic type system and automatic memory management. It supports multiple programming paradigms, including object-oriented, imperative, functional and procedural, and has a large and comprehensive standard library.^[29]

Python interpreters are available for many operating systems. CPython, the reference implementation of Python, is open sourcesoftware^[30] and has a community-based development model, as do nearly all of its variant implementations. CPython is managed by the non-profit Python Software Foundation.

From official documentation:
Python is powerful... and fast; 
plays well with others; 
runs everywhere; 
is friendly & easy to learn; 
is Open.

Installation

Installation process may vary depending on operational system: it can be installation on windows or just execution of "apt-get" in linux, but anyway, on official site you can find any information you need. 

Python shell

Easiest way to play with python - command shell, which can be open by running "python" command:

demien$ python
Python 2.7.13 (default, Nov 24 2017, 17:33:09)
[GCC 6.3.0 20170516] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>>

Now we can output standard "hello world" greeting message: 

>>> print("hello world")

hello world

Variables can be created by simple "=" operator: 

>>> name = "Huan Sebastyan"

>>> print("hello, {0} !!!".format(name))

hello, Huan Sebastyan !!!

PY Files

For long programs, python shell is not an option: it's better to put program content into file(or files) with extension ".py"  and run it by executing: python myprogram.py.  In code examples below I'm writing the code into file test.py  and executing it by running: python test.py. 

The main issue with Python files are the "blocks" of code : Python doesn't have {} or begin end operators to define scope(beginning and ending) of block, function or class. For this purpose Python use spaces: 

operator1

operator2 

def myFunction:

    function operator1

    function operator2

operator 3

Conditions in Python

Conditions are written without any bracers and of course for block inside condition we should use spaces to define begin and end:  

name = raw_input("What is your name? ")

greeting =""

if name == "Huan Sebastyan":

greeting = "Buenos dias"

else:

greeting = "Hello"

print("{0}, {1} !!!".format(greeting, name))

execution:

demien$ python test.py

What is your name? Joe

Hello, Joe !!!

demien$ python test.py

What is your name? Huan Sebastyan

Buenos dias, Huan Sebastyan !!!

Loops

There several types of loops in python: while and for

While loop:

value=""

while value!="end":

value = raw_input("Enter 'end' to quit or anything else to continue: ")

print("you entered:{0}".format(value))

print("you did it!")

execution: 

demien$ python test.py

Enter 'end' to quit or anything else to continue: hello

you entered:hello

Enter 'end' to quit or anything else to continue: world

you entered:world

Enter 'end' to quit or anything else to continue: end

you entered:end

you did it!

For loop:

count = int(raw_input("enter iteration count: "))

for i in range(1, count+1):

s = ""

for j in range(1, i+1):

s+="*"

print(s)

execution:

demien$ python test.py

enter iteration count: 5

*

**

***

****

*****

Functions

Definition begins with keyword "def" , bracers used for parameters: 

x = 10

y = 20

def test(param):

global x

y = 20

x=x+1

y=y+1

print("from function context: param={0}, x={1}, y={2}".format(param, x, y))

return "BYE!"

result = test("HELLO!")

print("from global context: result={0}, x={1}, y={2}".format(result, x, y))

Execution:

demien$ python test.py

from function context: param=HELLO!, x=11, y=21

from global context: result=BYE!, x=11, y=20

Modules

For big programs it's impossible to keep all code in one file. Python provides concept MODULE for this: 

Let's create a simple module in subfolder "tools":

tools/simple.py: 

def sayHi(name):

print("Hi, {0}".format(name))

__version__ = "0.0.1"

If we want to use it, beside module itself we also need an empty file __init__.py in the same folder(in subfolder "tools"):

demien$ ls -la tools/*.py

-rw-r--r-- 1 demien demien  0 mar 11 14:49 tools/__init__.py

-rw-r--r-- 1 demien demien 78 mar 11 14:46 tools/simple.py

Now we can use this module from our main program: 

import tools.simple as simple

simple.sayHi("Joe")

Execution: 

demien$ python test.py

Hi, Joe

Sometimes we may need to understand if the code is running by "direct" execution, or by importing it as module. For this we can check condition: if __name__ == "__main__":

- if it returns true, the code is being executed in "direct" mode. Our current module by direct execution produce no output. Let's update our module to produce some output, but only in case of "direct" execution: 

def sayHi(name):

print("Hi, {0}".format(name))

__version__ = "0.0.1"

if __name__ == "__main__":

sayHi("Huan Sebastyan")

So, now we can run our module in a direct way: 

demien$ python tools/simple.py

Hi, Huan Sebastyan

And now let's use it as module: 

demien$ python test.py

Hi, Joe

- output remained the same. 

Dir function

This function is showing "content" of variables and methods defined in module: 

>>> import tools.simple

>>> dir(tools.simple)

['__builtins__', '__doc__', '__file__', '__name__', '__package__', '__version__', 'sayHi']

Collection classes

there several collection classes in Python: lists, dictionary, set, tuples

Lists

It's a mutable structures for storing data in arrays.

colors = ["red", "black", "white"]

print "there are ",len(colors)," colors in my list:"

for color in colors:

print(color)

colors.append("blue")

colors.append("green")

print "few more there added, now it's ",len(colors)," of them"

colors.sort()

print("sorted:", colors)

del colors[0]

del colors[0]

print("first 2 were deleted:", colors)

Execution: 

demien$ python test.py

there are  3  colors in my list:

red

black

white

few more there added, now it's  5  of them

('sorted:', ['black', 'blue', 'green', 'red', 'white'])

('first 2 were deleted:', ['green', 'red', 'white'])

Tuples

Similar to lists  but they are immutable

answer = ("yes", "no")

print answer[0]

print answer[1]

execution:

demien$ python test.py

yes

no

Dictionary

It's structure like Map or Associated array. 

user = {

"name" : "Joe",

"surname" : "Black",

"address" : {

"country" : "USA",

"city" : "Houston"

}

}

print user["name"]

print user["address"]

print user["address"]["city"]

print "full list of pairs[key,value] in user dictionary:"

for key, value in user.items():

print "key=", key, " value=", value

Execution:

demien$ python test.py

Joe

{'country': 'USA', 'city': 'Houston'}

Houston

full list of pairs[key,value] in user dictionary:

key= surname  value= Black

key= name  value= Joe

key= address  value= {'country': 'USA', 'city': 'Houston'}

Sequence operations. 

Structures like lists, tuples and strings have list of common "sequence" operations.

colors = ["red", "black", "white", "blue", "gray", "green", "orange"]

colors.sort()

print colors

print('color 2 is', colors[2])

print('color -2 is', colors[-2])

print('colors 1 to 3 is', colors[1:3])

print('colors 2 to end is', colors[2:])

print('colors 1 to -1 is', colors[1:-1])

print('colors start to end is', colors[:])

Execution:

demien$ python test.py

['black', 'blue', 'gray', 'green', 'orange', 'red', 'white']

('color 2 is', 'gray')

('color -2 is', 'red')

('colors 1 to 3 is', ['blue', 'gray'])

('colors 2 to end is', ['gray', 'green', 'orange', 'red', 'white'])

('colors 1 to -1 is', ['blue', 'gray', 'green', 'orange', 'red'])

('colors start to end is', ['black', 'blue', 'gray', 'green', 'orange', 'red', 'white'])

Set

Another collection structure is Set. On sets we can apply some math logic, like AND, OR, XOR: 

colors1 = set(["red", "black", "white"])

colors2 = set(["white", "blue", "gray"])

print(colors1)

print(colors2)

print("& : ", colors1 & colors2)

print("| : ", colors1 | colors2)

print("^ : ", colors1 ^ colors2)

Execution: 

demien$ python test.py

set(['white', 'black', 'red'])

set(['blue', 'gray', 'white'])

('& : ', set(['white']))

('| : ', set(['blue', 'gray', 'black', 'white', 'red']))

('^ : ', set(['blue', 'gray', 'black', 'red']))

Classes

Main issues with  OOP in Python are:

- classes are being created without "new" keyword:  userJoe = User("Joe", "Black")

 - constructor has name: __init__

 - variables defined in class definition are class(not object) variables and should be accessed by className.variableName

 - "self" stands for "this" variable, and used for definition of object variables: self.name = name

-  methods which are using object variables have to explicitly define self as first input parameter: def sayHi(self)

- to define subclass, superclass name has to be passed as a "parameter" for subclass name: class AdminUser(User)

- to call superclass method (even constructor) format should be SuperClassName.SuperClassMethod: User.__init__(self, name, surname)

Example: 

class User:

userCount = 0

def __init__(self, name, surname):

self.name = name

self.surname = surname

User.userCount+=1

print("User #{0} was created!".format(User.userCount))

def sayHi(self):

print("Hi, I'm {0} {1}".format(self.name, self.surname) )

class AdminUser(User):

def __init__(self, name, surname, role):

User.__init__(self, name, surname)

self.role = role

def sayHi(self):

User.sayHi(self)

print(" and I'm the {0} !!!".format(self.role))

userJoe = User("Joe", "Black")

userJoe.sayHi()

userHuan = User("Huan", "Seastyan")

userHuan.sayHi()

userAdmin = AdminUser("Super", "Admin", "boss")

userAdmin.sayHi()

Execution: 

demien$ python test.py

User #1 was created!

Hi, I'm Joe Black

User #2 was created!

Hi, I'm Huan Seastyan

User #3 was created!

Hi, I'm Super Admin

  and I'm the boss !!!

Files

For working with files all we need is "file()" operation which takes parameters : 

1. file name

2. mode: read or write 

Let's create simple file in.txt with 2 lines of text: 

it's a test file

just as example

Now let's create a simple program which will convert this file to uppercase and write it into "out.txt" file: 

inFile = file("in.txt", "r")

outFile = file("out.txt", "w")

eof = False

while eof == False :

line = inFile.readline()

if len(line)==0:

eof = True

else:

outFile.write(line.upper())

inFile.close()

outFile.close()

Execution result: 

out.txt:

IT'S A TEST FILE

JUST AS EXAMPLE

Pickle

Module pickele (which should be imported) provides ability to save the object into file in "serialized" format. And, of course, later we can read it and deserialize back to original object. 

import pickle

class User:

userCount = 0

def __init__(self, name, surname):

self.name = name

self.surname = surname

User.userCount+=1

print("User #{0} was created!".format(User.userCount))

def sayHi(self):

print("Hi, I'm {0} {1}".format(self.name, self.surname) )

userJoe = User("Joe", "Black")

fout = open("joe.bak", "wb")

pickle.dump(userJoe, fout)

fout.close()

del(userJoe)

fin = open("joe.bak", "rb")

restored = pickle.load(fin)

print(restored)

restored.sayHi()

fin.close()

Execution: 

demien$ python test.py

User #1 was created!

<__main__.User instance at 0x7f05c0296ef0>

Hi, I'm Joe Black

Exceptions

Python has similar to other languages system of error handling with TRY, EXCEPT (which stands for CATCH) and FINALLY. On next example we are handling keyboard input  exceptions such as pressing Ctrl+C during input and raising our own exception if input length is less than expected: 

class ShortInputException(Exception):

"""A user-defined exception class."""

def __init__(self, length, atleast):

Exception.__init__(self)

self.length = length

self.atleast = atleast

try:

text = raw_input("Enter something ....")

if len(text) < 3 :

raise ShortInputException(len(text), 3)

except EOFError:

print("Why did you do an EOF on me?")

except KeyboardInterrupt:

print("You cancelled the operation.")

except ShortInputException as ex:

print(("ShortInputException: The input was " +

"{0} long, expected at least {1}")

.format(ex.length, ex.atleast))

else:

print("You entered {}".format(text))

finally:

print("done")

Execution: 

demien$ python test.py

Enter something ....12345

You entered 12345

done

demien$ python test.py

Enter something ....12

ShortInputException: The input was 2 long, expected at least 3

done

demien$ python test.py

Enter something ....^CYou cancelled the operation.

done

Try with resource: WITH

If we are opening something in TRY block, very often we have to close it in a FINALLY block (try with resource). To make this "automatically" we can use WITH construction: opened resource will be closed automatically: 

with open("in.txt") as fin:

for line in fin:

print(line)

The end

As for me, python has a lot of common with javascript: dynamic typing, inheritance, but more focused on "back-end" development of scripting. It's very simple but powerful.  Now a lot of big data frameworks are providing python api, so it's better to be familiar with this language.