Python Super Tutorial 2

super() 

• will allow us not to call explicitly
• Enable multiple inheritance

class Computer():

    def __init__(self, computer, ram, storage):

        self.computer = computer

        self.ram = ram

        self.storage = storage

# Class Mobile inherits Computer

class Mobile(Computer):

    def __init__(self, computer, ram, storage, model):

        super().__init__(computer, ram, storage)

        self.model = model

Apple = Mobile('Apple', 2, 64, 'iPhone X')

print('The mobile is:', Apple.computer)

print('The RAM is:', Apple.ram)

print('The storage is:', Apple.storage)

print('The model is:', Apple.model)

Python Unittest

import unittest

class TestStringMethods(unittest.TestCase):

    def test_upper(self):

         self.assertEqual('foo'.upper(), 'FOO')

    def test_isupper(self):

         self.assertTrue('FOO'.isupper())

         self.assertFalse('Foo'.isupper())

    def test_split(self):

         s = 'hello world'

         self.assertEqual(s.split(), ['hello', 'world'])

         # check that s.split fails when the separator is not a string

         with self.assertRaises(TypeError):

              s.split(2)

if __name__ == '__main__':

    unittest.main()

python collections Counter

from collections import Counter

listA = [40, 10, 20, 30, 10]
c = Counter(listA)

print(c)  #Counter({10: 2, 40: 1, 20: 1, 30: 1})

python collections OrderedDict

from collections import namedtuple
from collections import OrderedDict, defaultdict

# count no elements
# print dict in same order
d = OrderedDict() 
n = 5
ll = ['aaa', 'bbb', 'ccc', 'aaa']
for i in ll:
    d.setdefault(i, 0)  #it initialises to, else index error may come first time 
    d[i]+= 1

print(len(d))            # 3 
print(*d.values())   # 2 1 1
print(d)              #OrderedDict([('aaa', 2), ('bbb', 1), ('ccc', 1)])

Python Collections Counter, Defaultdict

1) Counter

from collections import Counter
myList = [10,10,20,30,4,5,3,2,3,4,2,1,2,30]
print(Counter(myList))
# Counter({2: 3, 10: 2, 30: 2, 4: 2, 3: 2, 20: 1, 5: 1, 1: 1})

print(Counter(myList).items())
# dict_items([(10, 2), (20, 1), (30, 2), (4, 2), (5, 1), (3, 2), (2, 3), (1, 1)])

print(Counter(myList).keys())
# dict_keys([10, 20, 30, 4, 5, 3, 2, 1])

print(Counter(myList).values())
# dict_values([2, 1, 2, 2, 1, 2, 3, 1])

2) defaultdict

from collections import defaultdict

a)

d = defaultdict(list)

# Even if key not exists it defaults to list/int
d['python'].append("awesome")
d['others'].append("not relevant")
d['python'].append("language")
d['test']

for i in d.items():
    print(i)

# O/P:
# ('python', ['awesome', 'language'])
# ('others', ['not relevant'])
# ('test', [])


b)

d = defaultdict(int)
d['without_val']
d['with_val'] = 100
for i in d.items():
    print(i)

# O/P:
# ('without_val', 0)
# ('with_val', 100)

c)

demo = defaultdict(int)

print(demo[300]) # 0

Git How to Revert a commit and push to new branch

Say you committed files to a wrong branch
How to revert and push new branch?

Existing Branch
git log --oneline
check for commit id: E.g., e0db5f7
git revert <commit_id>
git push

Create a new branch
git checkout -b new_branch
git checkout <commit_id> .    #this will copy commit files to local branch
git commit
git push

How to Configure Zookeeper and Kafka?

How to Configure Kafka?

# Download Kafka

• https://kafka.apache.org/downloads.html
• Download from Binary Downloads

# Kafka ENV  

• export KAFKA_HOME=$HOME/Workspace/prabhath/personal/kafka_2.12-2.5.0 
• export PATH=$KAFKA_HOME/bin:$PATH

Zookeeper config:

• bin/zookeeper-server-start.sh
• bin/zookeeper-server-stop.sh
• config/zookeeper.properties --> Default port: 2181, dataDir: /tmp/zookeeper

Kafka Config:

• bin/kafka-server-start.sh
• bin/kafka-server-stop.sh
• config/server.properties --> Default port: 9092

1) Start zookeeper

• zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties

2) Start Kafka server

• kafka-server-start.sh $KAFKA_HOME/config/server.properties

3) Create a Kafka topic

• kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic first_kafka_topic
• kafka-topics.sh --list --zookeeper localhost:2181 consumer_offsets
• It lists first_kafka_topic

4) Start Kafka Producer

• kafka-console-producer.sh --broker-list localhost:9092 --topic first_kafka_topic
• <start typing data>

5) Start Kafka Consumer

• kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first_kafka_topic --from-beginning

Python collections namedtuple

import collections

fields = ['OBJECTID', 'Identifier', 'Occurrence_Date', 'Day_of_Week', 'Occurrence_Month', 'Occurrence_Day', 'Occurrence_Year', 'Occurrence_Hour', 'CompStat_Month', 'CompStat_Day', 'CompStat_Year', 'Offense', 'Offense_Classification', 'Sector', 'Precinct', 'Borough', 'Jurisdiction', 'XCoordinate', 'YCoordinate', 'Location_1']

Crime = collections.namedtuple('Crime', fields)

row1_value = ['1', 'f070032d', '09/06/1940 07:30:00 PM', 'Friday', 'Sep', '6', '1940', '19', '9', '7', '2010', 'BURGLARY', 'FELONY', 'D', '66', 'BROOKLYN', 'N.Y. POLICE DEPT', '987478', '166141', '(40.6227027620001, -73.9883732929999)']

row1_obj = Crime(*row1_value)

print(row1_obj)


Output:
Crime(OBJECTID='1', Identifier='f070032d', Occurrence_Date='09/06/1940 07:30:0
0 PM', Day_of_Week='Friday', Occurrence_Month='Sep', Occurrence_Day='6', Occur
rence_Year='1940', Occurrence_Hour='19', CompStat_Month='9', CompStat_Day='7',
 CompStat_Year='2010', Offense='BURGLARY', Offense_Classification='FELONY', Se
ctor='D', Precinct='66', Borough='BROOKLYN', Jurisdiction='N.Y. POLICE DEPT', 
XCoordinate='987478', YCoordinate='166141', Location_1='(40.6227027620001, -73
.9883732929999)')

Python Puzzle Remove even numbers

Python Puzzle Remove even numbers

# Wrong approach (incorrect - using For loop)

def removeEven(List):

    print(id(List)) # 139909029905664

    for each in List:

       i f each % 2 == 0:

          List.remove(each)

myList = [152, 168, 154, 32, -55, 81, 146, -34, -124, -9, 4, -31, -131, -86, -190, -38]

print(id(myList)) # 139909029905664

print(myList) # [1, 2, 4, 5, 10, 6, 3]

removeEven(myList)

print(myList) # [168, 32, -55, 81, -34, -9, -31, -131, -190]

# Wrong as when element gets deleted, index goes down

print('-' * 60)

# Correct approach (using While loop)

def removeEvenNew(List):

    print(id(List))

    i = 0

    while i < len(List):

      if List[i] % 2 == 0:

          List.remove(List[i])

      else:

          i += 1 

myList = [152, 168, 154, 32, -55, 81, 146, -34, -124, -9, 4, -31, -131, -86, -190, -38]

print(id(myList))

print(myList) # 

removeEvenNew(myList)

print(myList) # [-55, 81, -9, -31, -131]

Output:

140407115793920

[152, 168, 154, 32, -55, 81, 146, -34, -124, -9, 4, -31, -131, -86, -190, -38]

140407115793920

[168, 32, -55, 81, -34, -9, -31, -131, -190]

------------------------------------------------------------

140407114861888

[152, 168, 154, 32, -55, 81, 146, -34, -124, -9, 4, -31, -131, -86, -190, -38]

140407114861888

[-55, 81, -9, -31, -131]

Python Lists Advanced

ll = [10, 20, 30, 40, 50]

# insert, remove, pop
ll.remove(20) #[10, 30, 40, 50]
ll.pop() # #[10, 30, 40]

ll = [1, 3, 5, 'seven']
ll.insert(0, 2) 
print(ll) # [2, 1, 3, 5, 'seven']

ll.pop(2) # pops 2nd index element
print(ll) # [2, 1, 5, 'seven']

ll.pop() # takes out last item
print(ll) # [2, 1, 5]

# Slice
gg = [1, 3, 5, 'seven', 'eight', 'nine', [10, 20,]]
print(gg[1:4])  # [3, 5, 'seven']
print(gg[3:])  # ['seven', 'eight', 'nine', [10, 20]]
print(gg[:3])  # [1, 3, 5]
print(gg[:])  # [1, 3, 5, 'seven', 'eight', 'nine', [10, 20]]

print(gg[-1:]) # [[10, 20]]
print(gg[:-1]) # [1, 3, 5, 'seven', 'eight', 'nine']

print(gg[-3:-1]) # ['eight', 'nine']

# list[start:stop:step]
print(gg[0:7:2]) # [1, 5, 'eight', [10, 20]]

ff = [1, 3, 5, 'seven', 'eight', 'nine']
print(ff) #[1, 3, 5, 'seven', 'eight', 'nine']
ff[2:2] = ['test']
print(ff) # [1, 3, 'test', 5, 'seven', 'eight', 'nine']
ff[1:3] = []
print(ff) # [1, 5, 'seven', 'eight', 'nine']

del ff[::2] # Delete even numbred indeces
print(ff)  # [5, 'eight']

# Concatenate
kk = [1, 2, 3, 4] # [1, 2, 3, 4]
kk += 'ab' # since string, it takes as two elements
print(kk) # [1, 2, 3, 4, 'a', 'b']

kk += ['c', 'd']
print(kk) # [1, 2, 3, 4, 'a', 'b', 'c', 'd']

kk.extend(['e', 'f'])
print(kk) # [1, 2, 3, 4, 'a', 'b', 'c', 'd', 'e', 'f']

# List Vs Array
# Array has homogenous elements
# Python arrays are just wrappers for C language
import array
# type: 'd' (float), initializer list: [1, 2, 3]
newArray = array.array('i', [1, 2, 3])
print(newArray) # array('i', [1, 2, 3])

Python AsyncIO Example

import asyncio
import aiohttp
import time

async def crawl_one_url(url, session):
    get_request = session.get(url)
    print(url)
    res = await get_request
    txt = await res.text()
    get_request.close()
    return txt


async def crawl_urls(urls_to_crawl):
    session = aiohttp.ClientSession()

    work_to_do = list()
    for url in urls_to_crawl:
        work_to_do.append(crawl_one_url(url, session))
    print(*work_to_do)
    res = await asyncio.gather(*work_to_do)
    # print(res)
    await session.close()
    return res


def main():
    t0 = time.time()
    urls_to_crawl = list()
    urls_to_crawl.append('http://blog.prabhathkota.com/search/label/python')
    urls_to_crawl.append('http://blog.prabhathkota.com/search/label/perl')
    urls_to_crawl.append('http://blog.prabhathkota.com/search/label/unix')
    urls_to_crawl.append('http://blog.prabhathkota.com/search/label/aws')
    urls_to_crawl.append('http://blog.prabhathkota.com/search/label/java')
    asyncio.run(crawl_urls(urls_to_crawl))
    elapsed = time.time() - t0
    print(f"{len(urls_to_crawl)} URLS downloaded in {elapsed:.2f}")


if __name__ == '__main__':
    main()


Output:
<coroutine object crawl_one_url at 0x7f5fe36181c0> <coroutine object crawl_on
e_url at 0x7f5fe3618240> <coroutine object crawl_one_url at 0x7f5fe36182c0> <
coroutine object crawl_one_url at 0x7f5fe3618340> <coroutine object crawl_one
_url at 0x7f5fe36183c0>
http://blog.prabhathkota.com/search/label/python
http://blog.prabhathkota.com/search/label/perl
http://blog.prabhathkota.com/search/label/unix
http://blog.prabhathkota.com/search/label/aws
http://blog.prabhathkota.com/search/label/java
5 URLS downloaded in 0.48

Python Generator Send & Recieve

#############################################
# Yield returns a generator object
# use next() to access the return value
# If you attempt to invoke next() on a generator object that had already produced (yielded) all its values, it will throw StopIteration exception
# We can pass data to a generator function using the send() method defined
# Use generators to generate values and coroutines to consume values. Generator functions receive values are called coroutines. 

#############################################


############# Test1 (send & receive) ############ 
def test_yield_send():
    while True:
        item = yield
        print(f'Received item: {item}')

if __name__ == '__main__':
    gen = test_yield_send()
    print(gen)  #<generator object test_yield at 0x7ff2f0510c80>
    next(gen)
    gen.send(100)

Output:

<generator object test_yield_send at 0x7ff700b45c80>

Received item: 100

############ Test2 (send & receive) ############ 
def generate_num():
    i = 0
    while True:
        i += 1
        t = (yield i)
        print(t)


if __name__ == "__main__":
    gen = generate_num()

    item = gen.send(None)
    print("First received " + str(item))

    for i in range(0, 5):
        item = gen.send(100 + i)
        print("Other received " + str(item))


Output:
First received 1
100
Other received 2
101
Other received 3
102
Other received 4
103
Other received 5
104
Other received 6

Python Generator yield

"""
Yield returns a generator object
use next() to access the return value
If you attempt to invoke next() on a generator object that had already produced (yielded) all its values, it will throw StopIteration exception
It can yield multiple items as well (as shown in second example)

"""

################# Test1 #################
def test_yield():
    yield 'hi'

if __name__ == '__main__':
  a = test_yield()
  print(a)  #<generator object test_yield at 0x7ff2f0510c80>
  val = next(a)
  print(val) # hi

Output:

hi

################# Test2 #################
# You can yield multiple values
def test_yield(name):
    yield 'hi'
    yield name

if __name__ == '__main__':
  a = test_yield('John')
  print(a)  #<generator object test_yield at 0x7ff2f0510c85>
  for item in a:
    print(item)

Output:

<generator object test_yield at 0x7f93f8066c80>

hi

John

Python multiprocessing spawn

Fork Vs Spawn
Fork child inherit all resources from the parent process

Fork is the default method for multi-processing
Spawn child doesn't inherit any resources from the parent process other than those required to execute the specified callable target.
Spawning a process is slower (because it re-imports) than forking a process

Spawn
Spawn is essentially a combination of fork followed by an exec system call
When a child process is spawned, anything imported at the module level in the (above )__main__ module of the parent process gets reimported in the child
Anything below __main__ will not be copied

test1.py
#########
Inside test1

test1.py
#########
Inside test2

spawn_test.py
############
from multiprocessing import Process
import test1
import test2

def process_task():
    print("I am inside child process")


if __name__ == '__main__':

    # Change the method to 'spawn' and verify
    # that the modules are reimported in the child process

    ### Spawn
    multiprocessing.set_start_method('spawn') 
    process = Process(target=process_task)
    process.start()
    process.join()
    print("I am inside parent process")


# Child reimports the test1, test2 module again...

Output:
Inside test1
Inside test2
Inside test1
Inside test2
I am child process
I am parent process

Python multiprocessing fork

Fork
# When we fork, the entire Python process is duplicated in memory including the Python interpreter, code, libraries, current stack, etc.
# This creates a new copy of the python interpreter.
# Fork creates two python interpreters each with its own GIL.
# Fork is faster than Spawn (Fork child inherit all resources from the parent process, Spawn re-imports all above main() method)

# Fork is the default method for multi-processing

 
#Disadvantages of Fork
# It won't work on windows
# When child shares parent libraries, values, data-structures, if a lock acquired by parent, child ends up waiting for that lock ever
# Very hard to debug when you import a third-party module/library that uses threads behind the scenes
# Fork and Multi-threading won't go well

from multiprocessing import Process
import multiprocessing
import os

file_desc = None

def process_task1():
    # write to the file in child process
    file_desc.write(f"\nWritten by child process with id {os.getpid()}")
    file_desc.flush()


if __name__ == '__main__':
    # create a file in the parent process
    file_desc = open("sample.txt", "w")
    file_desc.write(f"\nWritten by parent process with id {os.getpid()}")
    file_desc.flush()

    # Fork is default method to create a process
    # multiprocessing.set_start_method('fork')

    p = Process(target=process_task1)
    p.start()
    p.join()
    file_desc.close()

    file_des = open("sample.txt", "r")
    print(file_des.read())

    os.remove("sample.txt")


Output:
Written by parent process with id 288
Written by child process with id 294

python how to overcome GIL

Ref: https://realpython.com/python-gil/

# single_threaded.py
import time
from threading import Thread

COUNT = 50000000

def countdown(n):
    while n>0:
        n -= 1

start = time.time()
countdown(COUNT)
end = time.time()

print('Time taken in seconds -', end - start)
#Time taken in seconds - 6.20024037361145


# multi_threaded.py
import time
from threading import Thread

COUNT = 50000000

def countdown(n):
    while n>0:
        n -= 1

t1 = Thread(target=countdown, args=(COUNT//2,))
t2 = Thread(target=countdown, args=(COUNT//2,))

start = time.time()
t1.start()
t2.start()
t1.join()
t2.join()
end = time.time()

print('Time taken in seconds -', end - start)
# Time taken in seconds - 6.924342632293701
# No improvement here due to GIL & due to sharing of lock & releases


# multi_processing.py
# Uses cores instead of threads
from multiprocessing import Pool
import time

COUNT = 50000000
def countdown(n):
    while n>0:
        n -= 1

if __name__ == '__main__':
    pool = Pool(processes=2)
    start = time.time()
    r1 = pool.apply_async(countdown, [COUNT//2])
    r2 = pool.apply_async(countdown, [COUNT//2])
    pool.close()
    pool.join()
    end = time.time()
    print('Time taken in seconds -', end - start)
    #Time taken in seconds - 4.060242414474487

Python GIL, lock, mutex, semaphore

Lock
#####
A lock allows only one thread to enter the part that's locked and the lock is not shared with any other processes.

Mutex
#######
Mutex as the name means mutual exclusion.
A mutex is used to guard shared data such as a list, dict.
A mutex allows only a single thread to access a resource or critical section.
A mutex is the same as a lock but it can be system wide (shared by multiple processes).
Same thread can acquire and release lock
Owned by thread

Semaphore
##########
Semaphore, is used for limiting access to a collection of resources.
A semaphore does the same as a mutex but allows x number of threads to enter, this can be used
E.g., to limit the number of cpu, io or ram intensive tasks running at the same time.
      Pool of DB connections to be handed out to requesting threads
Different threads can call acquire and release on the semaphore
No ownership

GIL
####
Ref: https://realpython.com/python-gil/

The Python interpreter can only execute a single thread at a time
If your machine has one or hundred processors, the Python interpreter is only able to run a single thread at a time using a single processor.
Two threads on a machine with two available processors can't be executed in parallel each running on a single CPU.
This lock is known as the Global Interpreter Lock using CPython (to avoid deadlocks)
Python implementations which overcome the GIL altogether. E.g., Jython, IronPython and pypy-stm.

Python uses reference counting for memory management. (unlike garbage collection in other languages)
It means that objects created in Python have a reference count variable that keeps track of the number of references that point to the object. When this count reaches zero, the memory occupied by the object is released.

This reference count variable can be kept safe by adding locks to all data structures that are shared across threads so that they are not modified inconsistently.

But adding a lock to each object or groups of objects means multiple locks will exist which can cause another problem—Deadlocks (deadlocks can only happen if there is more than one lock). Another side effect would be decreased performance caused by the repeated acquisition and release of locks.

The GIL is a single lock on the interpreter itself which adds a rule that execution of any Python bytecode requires acquiring the interpreter lock. This prevents deadlocks (as there is only one lock) and doesn’t introduce much performance overhead. But it effectively makes any CPU-bound Python program single-threaded.

Python sum using multiprocessing pool

#####
# Read all text files & calculate Sum of all numbers in txt files

# Threading/input1.txt
# 100
# 200
# 300
# 400
# 500

# Threading/input2.txt
# 600
# 700
# 800
# 900
# 1000
#####

import glob
import multiprocessing

def get_txt_files():
    my_files = []
    for each in glob.glob('Threading/*.txt'):
        my_files.append(each)
    return my_files

def calculate_sum(input_file):
    with open(input_file) as fh:
      return sum(map(int, fh.readlines()))


if __name__ == '__main__':
    my_files = get_txt_files()
    #['Threading/input1.txt', 'Threading/input2.txt']
    print(my_files)

    # sys.exit()
    pool = multiprocessing.Pool(processes=2)
    results = pool.map(calculate_sum, my_files)
    print(f'Sum of individual files: {results}')
    total_sum = sum(results)
    print(f'Total Sum: {total_sum}')


Output:
######
['Threading/input1.txt', 'Threading/input2.txt']
Sum of individual files: [1500, 4000]
Total Sum: 5500

Python thread lock release

#####
#If you don't use lock, sum will be not printed as 500000
#If you use lock, you get the sum as expected 500000 (5 threadsList, 100000 each)
#####

from threading import Thread
from threading import Lock
import sys

class CounterClass:
    def __init__(self):
        self.count = 0
        self.lock = Lock()

    def increment(self):
        for _ in range(100000):
            self.lock.acquire()
            self.count += 1
            self.lock.release()


if __name__ == "__main__":
   
    # Sets the thread switch interval
    sys.setswitchinterval(0.005)

    numThreads = 5
    threadsList = []
    counterObj = CounterClass()

    for i in range(0, numThreads):
        threadsList.append(Thread(target=counterObj.increment))

    for i in range(0, numThreads):
        threadsList[i].start()

    for i in range(0, numThreads):
        threadsList[i].join()

    if counterObj.count != 500000:
        print(" count = {0}".format(counterObj.count))
    else:
        print(" count = 500000")

Python Palindrome

Using Recursion

def isPalindrome(testVariable):
  print(testVariable)
  if len(testVariable) <= 1:
    return True
  
  length = len(testVariable)
  if testVariable[0] == testVariable[length-1]:
      return isPalindrome(testVariable[1:length-1])

  return False

a = isPalindrome('MADAM')
print(a)  ## True

Using normal way

def isPalindrome(testVariable):
  if testVariable == testVariable[::-1]:
    return True
  return False

a = isPalindrome('MADAM')
print(a)  ## True

Python recursion

def factorial(num):
    if num == 1:
       return num
    else:
       return num * factorial(num-1)

target = factorial(5)
print(target)  ## 120


def square(num):
    if num == 1:
      return num
    else:
      return square(num-1) + 2*num - 1

target = square(6)

print(target)  ## 36

python reverse string

def reverse_fuc(input_str):
    reverse = ''
    length = len(input_str) - 1
    while length >= 0:
       reverse = reverse + input_str[length]
       length = length - 1
return reverse

target = reverse_fuc('prabhath')
print(target)  #htahbarp

Pyspark Tutorial 1

#............................................................................
##   Loading and Storing Data
#............................................................................

#Create a Spark Session
SpSession = SparkSession \
    .builder \
    .master("local[2]") \ #[2] is no of partitions
    .appName("learnSpark") \
    .config("spark.executor.memory", "1g") \
    .config("spark.cores.max","2") \
    .getOrCreate()

#Get the Spark Context from Spark Session
SpContext = SpSession.sparkContext

#Get the Spark Context from Spark Session
SpContext = SpSession.sparkContext
print SpContext

#Create an RDD by loading from a file
tweetsRDD = SpContext.textFile("movietweets.csv")
print tweetsRDD

#Action - Count the number of tweets
print tweetsRDD.count()

#show top 5 records
print tweetsRDD.take(5)

#Transform Data - change to upper Case
ucRDD = tweetsRDD.map( lambda x : x.upper() )
print ucRDD.take(5)

#Load from a collection
collData = SpContext.parallelize([4,3,8,5,8])
collData.collect()  # bring the entire RDD to the driver node, could be expensive

#Load the file. Lazy initialization
autoData = SpContext.textFile("auto-data.csv")
autoData.cache()
#Loads only now.
print autoData.count() #198
print autoData.first() #prints 1st line
print autoData.take(5) #gives you a list of 5 elements

#Save to a local file. First collect the RDD to the master
#and then save as local file.
autoDataFile = open("auto-data-saved.csv","w")
autoDataFile.write("\n".join(autoData.collect()))
autoDataFile.close()

#............................................................................
##   Transformations
#............................................................................

#Map and create a new RDD
tsvData=autoData.map(lambda x : x.replace(",","\t"))
print tsvData.take(5)

#Filter and create a new RDD
toyotaData=autoData.filter(lambda x: "toyota" in x)
print toyotaData #RDD object it prints
print toyotaData.count() #32

#FlatMap
words=toyotaData.flatMap(lambda line: line.split(","))
print words.count() #384 words in 32 lines of toyota
print words.take(20) #it takes 20 words from 384 and print
print '----'    

#Distinct
for numbData in collData.distinct().collect():
    print(numbData)
print '----'    

#Set operations
words1 = SpContext.parallelize(["hello","war","peace","world"])
words2 = SpContext.parallelize(["war","peace","universe"])

for unions in words1.union(words2).distinct().collect():
    print(unions)

print '----'    
for intersects in words1.intersection(words2).collect():
    print(intersects)

print '----'    
#Using functions for transformation
#cleanse and transform an RDD
def cleanseRDD(autoStr) :
    if isinstance(autoStr, int) :
        return autoStr
    attList=autoStr.split(",")
    #convert doors to a number str
    if attList[3] == "two" :
         attList[3]="2"
    else :
         attList[3]="4"
    #Convert Drive to uppercase
    attList[5] = attList[5].upper()
    return ",".join(attList)
    
cleanedData=autoData.map(cleanseRDD)
print cleanedData.collect() #it returns a List

#............................................................................
##   Actions
#............................................................................

#reduce - compute the sum
collData.collect()  #[4,3,8,5,8]
print collData.reduce(lambda x,y: x+y) #28

print '----'    
#find the shortest line - reduce() RDD function 
print autoData.reduce(lambda x,y: x if len(x) < len(y) else y)

print '----'    

#Use a function to perform reduce 
def getMPG( autoStr) :
    if isinstance(autoStr, int) :
        return autoStr
    attList=autoStr.split(",")
    if attList[9].isdigit() :
        return int(attList[9])
    else:
        return 0

#find average MPG-City for all cars    
print autoData.reduce(lambda x,y : getMPG(x) + getMPG(y)) \
    / (autoData.count()-1.0)  # account for header line

#............................................................................
##   Working with Key/Value RDDs
#............................................................................

#create a Key Value RDD of auto Brand and Horsepower
cylData = autoData.map( lambda x: ( x.split(",")[0], \
    x.split(",")[7]))
print cylData.count() #198
print '----'    
print cylData.take(5) #take first 5 key-value pairs
print '----'    
print cylData.keys().collect() #get all keys from key-value pair RDD cylData
print '----'    

#Remove header row
header = cylData.first()
print header
cylHPData= cylData.filter(lambda line: line != header) #get data without header
print cylHPData.count() #197

print '----'    

#Find average HP by Brand
#Add a count 1 to each record and then reduce to find totals of HP and counts
addOne = cylHPData.mapValues(lambda x: (x, 1))
print addOne.collect() #(u'bmw', (u'182', 1)), (u'mercedes-benz', (u'123', 1))......, (u'bmw', (u'182', 1)), (u'mercedes-benz', (u'184', 1)) ]

print '----'    
brandValues= addOne \
    .reduceByKey(lambda x, y: (int(x[0]) + int(y[0]), x[1] + y[1])) 
print brandValues.collect() #(u'mercedes-benz', (1170, 8)), (u'mitsubishi', (1353, 13)), (u'saab', (760, 6)), (u'volkswagen', (973, 12))
print '----'    

#find average by dividing HP total by count total
print brandValues.mapValues(lambda x: int(x[0])/int(x[1])). \
    collect()
#[(u'dodge', 84), (u'mercury', 175), (u'jaguar', 204), (u'alfa-romero', 125), (u'nissan', 102), (u'toyota', 92), (u'plymouth', 86), (u'mazda', 86), (u'subaru', 86), (u'peugot', 99), (u'porsche', 191), (u'isuzu', 84), (u'chevrolet', 62), (u'honda', 80), (u'volvo', 128), (u'bmw', 138), (u'mercedes-benz', 146), (u'mitsubishi', 104), (u'saab', 126), (u'volkswagen', 81), (u'audi', 114)]

print '----'    

#............................................................................
##   Advanced Spark : Accumulators & Broadcast Variables
#............................................................................

#function that splits the line as well as counts sedans and hatchbacks
#Speed optimization

    
#Initialize accumulator
sedanCount = SpContext.accumulator(0)
hatchbackCount = SpContext.accumulator(0)

#Set Broadcast variable
sedanText=SpContext.broadcast("sedan")
hatchbackText=SpContext.broadcast("hatchback")

def splitLines(line) :
    global sedanCount
    global hatchbackCount

    #Use broadcast variable to do comparison and set accumulator
    if sedanText.value in line:
       sedanCount += 1
    if hatchbackText.value in line:
       hatchbackCount += 1
        
    return line.split(",")


#do the map
splitData=autoData.map(splitLines)

#Make it execute the map (lazy execution)
print splitData.count()
print '----'    
print(sedanCount, hatchbackCount)
print '----'    

#............................................................................
##   Advanced Spark : Partitions
#............................................................................
print collData.getNumPartitions()

#Specify no. of partitions.
collData = SpContext.parallelize([3,5,4,7,4], 4)
print collData.cache()
print collData.count()

print '----'    
print collData.getNumPartitions()

print '----'    
#localhost:4040 shows the current spark instance

RDD 

• Transformation - map
• Action - reduce, take
• lazy loads

Persist()

• percentofDelayedFlights = flightsParsed.filter(lambda x:x.delay>0).count() / flaot(flightsParsed.count()) 
• Here we are using flightsParsed twice, instead we can force RDD to be materialized once using persist()
• flightsParsed.persist()  # Cached, huge performance saving
• Any RDD you use over and over again, you need to persist() to improve performance
• flightsParsed.unpersist() # UnCache, free-up memory

aggregate()

• totalDistance = flightsParsed.filter(lambda x:x.distance).reduce(lambda x,y:x+y)
• avgDistance = totalDistance / flightsParsed.count()
• Instead of two actions reduce & count, we can go with one using aggregate
• 
  

Freq Dist by Hours

•  flightsParsed.filter(lambda x: int(x.dep_delay/60)).countByValue()

Alembic Basics

Alembic

alembic revision --autogenerate
alembic history

alembic stamp head
alembic stamp <rev_no>

alembic upgrade head
alembic downgrade base

pytest Tutorial 3 - pycharm options

• pytest.ini
• You can define default options like -v, -m, -x etc.,
• [pyest]
• addopts = -v
• Auto-run options
• When you make some changes, it auto-run
• You can set interval of 2, 3, 5 secs etc.,
• You can find this in settings gear icon in Pycharm
• Make default pytest
• By default it is pointed to unittest in Pycharm
• You can change to pytest in PyCharm test configurations under
• Tools -> Python Integrated Tools -> Default Test Runner
• Run configuration for individual files
• For each file, you can set -v, -m options
• Run -> Default Configurations -> Additional Arguments
• -v
• Set options Globally for all instead of individual files
• Run -> Edit Configurations -> Defaults -> Python Tests -> py.test
• Add it here in Additional Arguments
• This will apply to all the test files
• Window -> Edit Tabs -> Split Vertically
• You need to work test cases looking at your code files/Readme.
• This will help in view both files vertically at the same time

pytest in classes

import pytest

class TestSomeStuff():
    def test_one(self):
         assert 1 == 1
   
    def test_two(self):
         assert 2 == 2

    def test_three(self):
         assert 3 == 3

class TestOtherStuff():
    def test_eleven(self):
         assert 11 == 11
   
    def test_twenty_two(self):
         assert 22 == 22

    def test_thirty_three(self):
         assert 33 == 33

We can run as all the tests a different suites as classes.
TestSomeStuff
TestOtherStuff

pytest Tutorial 2 -Fixtures

student.py

import json

class StudentDB:
   def __init__(self):
        self.__data = None

   def connect(self, data_file):
        with open(data_file) as json_file:
            self.__data = json.loads(data_file)

   def get_data(self, name):
        for student in self.__data['students']:
             if student['name'] == name:
                 return student

test_student.py

from student import StudentDB
import pytest

# Fixture is more like a replacement for writing both setup and teardown
# If not fixture, you can write separate functions for setup_module, teardown_module
# You need to pass db to all the unit test functions
#If you not pass scope=module, this fixture is called for each and every unit test
@pytest.fixture(scope='module')  
def db:
    print('-------Inside setup-------')
    db = StudentDB()
    db.connect('data.json')
    yield db
    print('-------Inside teardown-------')  # Since this is module level, this is called at the end of module
    db.close()

def test_scott_data(db):  # Passed from above texture
    scott_data = db.get_data('Scott')
    assert scott_data['id'] == 1
    assert scott_data['name'] == 'Scott'
    assert scott_data['result'] == 'Pass'

def test_mark_data(db):  # Passed from above texture
    scott_data = db.get_data('Mark')
    assert scott_data['id'] == 2
    assert scott_data['name'] == 'Mark'
    assert scott_data['result'] == 'Fail'



 
 

pytest Tutorial 1 - Beginner

math_func.py

def add(x,y):
   return x+y

def product(x,y)
   return x*y

test_math_func.py

from math_func import add, subtract
import pytest
import sys

# @pytest.mark.skip(reason="Do not run number add test")
# @pytest.mark.skipif(sys.version_info < (3, 3), reason="Do not run number add test") 
@pytest.mark.number
def test_add():
   assert add(3,4) == 7
   print(f'Inside test_add: {assert add(3,4)}')

@pytest.mark.parameterize('arg1, arg2, result',
[
(7, 3, 10),
('Hello', ' World', 'Hello World')
(5.5, 4.5, 10)
]
)
def test_add_parameterize(arg1, arg2, result):
   assert add(arg1, arg2) == result

@pytest.mark.number
def test_product():
   assert add(3,4) == 12
 
@pytest.mark.strings
def test_add_strings():
   assert add(3,4) == 7
   result add('Hello', ' World')
   assert result == 'Hello World'
   assert 'Hello' in result
   assert type(result) is str 

@pytest.mark.strings
def test_product_strings():
   assert product('Hello', 3) == 'HelloHelloHello'

How to run
pytest test_match_func.py
pytest test_match_func.py -v

pytest test_match_func.py::test_add

(-k Expression)
pytest -v -k "add"  # runs tests with "add" string, run both test_add & test_add_strings
pytest -v -k "add or string" # runs all 3
pytest -v -k "add and string"  # runs test_add_strings alone

(-m, --markers)
pytest -v -m number # run those marked as number
pytest -v -m strings  # run those marked as strings

(-x, --exitfirst)

pytest -v -x  # If any first failure, it totally exits and does not execute following tests

pytest -v -x --tb=no  # It won't show error stack trace

pytest -v --maxfail=2

pytest -v -s  # -s option will output any print lines

pytest -v -q  # quite mode, will not print how many test passed

pytest Tutorial 2 -Fixtures (check the other blog)

Pandas Dataframes

# Info - lists all of our columns, also gives data types
df.info()

# Set options
pd.set_option('display.max_columns', 85)
pd.set_option('display.max_rows', 85)

# head
df.head(10)

# tail
df.tail(10)

# List all columns
df.columns

# Dataframe shape

print(df1.shape)

# Series - is a list / a column in a data frame
# A data frame is a container for multiple Series (columns)
people = {
              'first' : ['John', 'Mary', 'Linda'],
              'last' : ['Doe', 'Lee', 'Doe'],
              'email' : ['johndoe@gmail.com', 'marylee@gmail.com', 'lindadoe@gmail.com']
 }
df = pd.DataFrame(people)
type(df['email'])  # gives pandas.core.series.Series
df['email']  # gives list of emails
df.email  # gives same as above

pd.DataFrame.from_records (list of tuples)

list = [('John',30,'Bangalore'), ('Doe',25,'Chennai')] df = pd.DataFrame.from_records(list, columns=['Name', 'Age', 'City']) print(df.head(10))

df['email'] Vs df.email 

• df['email'] is better to use over df.email
• shape is an attribute of data frame, suppose you have a column name as shape in your DF
• df['shape'] will give you result of shape column list 
• df.shape will not work properly as you expect

# iloc Vs loc (Integer Location Vs Location)
# iloc (Search by position)
    df.iloc[0] # return a Series object for 0th index, access row with iloc
    df.iloc[ [0, 1] ] # returns first two rows of data, returns a Data Frame
    df.iloc[ [0, 1],  2] # returns email (2nd column) of first two rows
# loc  (Search by label) (row_index, column_indexer)

    df.loc[0] # Same as iloc, returns Series object
    df.loc[ [ 0, 1] ] # Same as iloc, returns data frame object
    df.loc[ [ 0, 1 ], 'email' ]  # Instead of index, you can use column name
    df.loc[ [ 0, 1 ], ['email', 'last_name'] ] # Filter by multiple column names for first two rows
    df.loc[ 0, 'Hobby']  # Value of Hobby column for first row
    df.loc[ 0:5, 'Hobby:Employment']  # Values of Hobby to Employment columns for first six rows, slicing is inclusive   # see df.columns to get all the columns
    df.loc[ df['email'] == 'abc@gmail.com',  : ]  # Rows of all with specific email (all columns)

# Get row by index
       Name  Age PinCode
0    Alex   10     500
1     Bob   12     600
2  Clarke   13     589

df.loc[2] # Gets object of Clarke by index
df1 = df.set_index('Name')  # Set Name as index, inplace=True is also there
print(df1)

             Age PinCode
Name
Alex     10     500
Bob      12     600
Clarke   13     589

df1.loc['Clarke']  # You can get the row by passing Name, since Name is the index


# Get unique values of Column (Yes/No)
df["gender"].value_counts()
Male 100
Female 200

# Unique

ll = ['John', 'Doe', 'Tom', 'Doe']

df1 = pd.Series(ll)

df2 = pd.unique(df1)

print(type(df2)) #<class 'numpy.ndarray'>

print(df2)         #['John' 'Doe' 'Tom']

# Filter

df1_link_keys = df1[df1.link_key != '']

df1_no_link_keys = df1[df1.link_key == '']

# Filter multiple/specific columns
df1 = df[['first', 'last', 'email']]  # Gives DF in return

# Filter isin

df1_subset = df1[df1.index.isin(100, 200)]

# Replace something with another

df1 = df1.replace('NULL', '')

df1 = df1.fillna('')

# Drop index

df1 = df1.drop(matched_index_list)

df1 = df1.drop(100)

# Column to list

df1_names_list = df1['name'].tolist() # column values to list

df1_index_list = df1.index.values.tolist() # index values to list

# Drop duplicates

df1.drop_duplicates(keep='last', inplace=True)  # Keep the last occurence

# Drop duplicates based on specific columns

df1.drop_duplicates(subset=['id', 'name', 'city', 'state'], keep='last', inplace=True)

df1.drop_duplicates(subset=['id', 'name', 'city', 'state'], keep='first', inplace=True)

df1 = df1.drop_duplicates(subset=['id', 'name', 'city', 'state'], keep='last')  # inplace is not true

df1 = df1.drop_duplicates(subset=['id', 'name', 'city', 'state'], keep='first') # inplace is not true

# Drop duplicates with state is empty 

# if only one record with state as empty, keep that

# If two same rows with one state is there & other state is empty, filter one with empty

s_maxes = df1.groupby(['id', 'name', 'city']).state.transform(max)

df1 = df1.loc[df1.state == s_maxes]

# Dataframe To Table

table_dtype = {

                'name': VARCHAR(),

                'city': VARCHAR(),

                'state': VARCHAR(),

                'pincode': INTEGER()

              }

df1.to_sql(con=self.db_conn, name=table_name, dtype=table_dtype,

           if_exists='replace', index=False)

# Dataframe To JSON

df.to_json('data_frame.json', orient='table')

# From Table to Dataframe

raw_data_sql = f""" select id, name, city, state

from event_raw_data

                """

df1 = pd.read_sql(raw_data_sql, con=self.db_conn, index_col="id")

# Loop Dataframe

for index, row in df1.iterrows():

print(index, row.name, row.city, row.state)

# How to subtract rows of one pandas data frame from another?

# You want to achieve df1 - df2

df1 = df1[~df1.index.isin(df2.index)]

# Read CSV
df1 = pd.read_csv('data/test.csv')
df1.shape  # (10, 5)

SQLAlchemy commands

SQLAlchemy
##########
from sqlalchemy import func
from sqlalchemy.orm import Session
from sqlalchemy.ext.automap import automap_base
from sqlalchemy import create_engine

self.db_engine, self.db_conn = create_db_conn()

self.Base = automap_base()
self.Base.prepare(self.db_engine, reflect=True)
self.db_session = Session(self.db_engine)

self.table1_obj = self.Base.classes.table1
self.table2_obj = self.Base.classes.table2

DATABASE_URI = 'postgresql://postgres:password@localhost:5432/test'

def create_db_conn():
    db_engine = None
    db_conn = None
    try:
        sql_alchemy_engine = create_engine(f'{DATABASE_URI}')

    return db_engine, db_conn



COUNT
#####
rows = vobj.db_session.query(self.table1_obj).count()
print(rows)
rows = vobj.db_session.query(self.table2_obj).count()
print(rows)

activities = self.db_session.query(self.table1_obj).order_by(
            self.table1_obj.id).all()

UPDATE
######
ea = self.db_session.query(self.table2_obj).get(self.activity_id)
            ea.status = status
            self.db_session.commit()

GET
####
current_activity = self.db_session.query(self.table1_obj).filter_by(
    activity_id=self.venue_base_activity_id,
    source_id=row['source_id'],
    venue_id=row['vid']).first()
if current_activity:
    link_key = current_activity.link_key
    market_place_df.loc[index, 'link_key'] = link_key

GET Specific Columns
####################
result = db_session.query(table1_obj.source_id, table1_obj.vid, table1_obj.name,
                              table1_obj.city, table1_obj.state)\
            .filter(table1_obj.source_id.in_((1, 2)))\
            .group_by(table1_obj.source_id, table1_obj.vid, table1_obj.name,
                      table1_obj.city, table1_obj.state)\
            .order_by(table1_obj.source_id).all()

for row in result:
    try:
        source_id = row[0]
        name = clean_string(row[2])
        city = clean_string(row[3])
        state = clean_string(row[4])

GET Count
#########
ws_count = self.db_session.query(func.count(self.table1_obj.id))\
                            .filter_by(source_id=market_place2_source_id).scalar()

group_by
#######
HygroRecord.query.group_by(HygroRecord.sensor_uuid)
        .having(func.max(HygroRecord.timestamp)).all())

insert
######
self.db_session.add(
            self.table1(
                id=row_id,
                name=name
            )
        )
self.db_session.commit()

Insert - bulk insert
#####################
ll = []
ll.append(self.table2(
        id=self.activity_id,
        name=name,
        city=city,
        state=state
    )
)
self.db_session.bulk_save_objects(ll)
self.db_session.commit()


flush
########

    f = vobj.table1(
            id1=9,
            id2=18
        )
    vobj.db_session.add(f)
    vobj.db_session.flush() #without commit if you want to get insert id
    print(f.id)
    vobj.db_session.commit()