sparkutils_for_erdal.py

# RUN THIS SSH:
# ssh imo2123 -tC "imo-pyspark --conf spark.ui.port=0 --num-executors 100"
# You can also change: --driver-memory 2g --executor-memory 2g
# You can increase --num-executors as well, but be careful during pick hours as you might wait a bit for resources from scheduler if you set too big
# AND THEN COPY FOLLOWING INTO PYSPARK SHELL:

import json
import datetime
import time

def format_line(line):
    try:
        splitted_line = line.split(" ", 1)
        t = float(splitted_line[0])
        data = json.loads(splitted_line[1])
        data['time'] = t
        return data
    except:
        return None

def get_fields(rdd):
    return b.flatMap(lambda j: j.keys()).distinct().collect()

def get_columns(rdd, col_names):
    return rdd.map(lambda dic_line:
                   {k: dic_line.get(k, None) for k in col_names})

def get_valid_lines_as_dict(rdd):
    return (rdd.map(format_line).filter(lambda dic_line: dic_line is not None))


def load_topic(topic, date=None, partition="*"):
    if date is None:
        date = datetime.datetime.now()
    else:
        date = datetime.datetime.strptime(date, "%Y-%m-%d")
    date = date.strftime("%Y/%m/%d")
    a = sc.textFile("hdfs:///data/logs/%s/%s/%s/*" % (topic, date, partition))
    return get_valid_lines_as_dict(a)


# -------------------------   QUICK TUTORIAL   --------------------------------
# group by (country_name, carrier_name) and count
# Load logs
a = load_topic("ImoAccount.registrations", date="2017-02-05")

# show one line to see how it looks
a.take(1)

# grouping:
# 1) creates (key, value) pairs, where key is whatever column, and value is 1
# 2) reduce by key to sum those "ones"
b = a.map(lambda x: ( (x['country_name'], x['carrier_name']), 1 ) ).reduceByKey(lambda acc, val: acc+val)

# show 3 the biggest
b.takeOrdered(3, lambda (k,v): -v)

# For more google: "Spark RDD transformations", e.g. https://www.supergloo.com/fieldnotes/apache-spark-examples-of-transformations/