<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 The Split-Apply-Combine Pattern in Data Science and Python
Tobias Brandt
<img src="img/argon_logo.png" align=left width=200>
<!-- <img src="http
Step1:  <pre>
{
    <b>"payload"
Step2:  Example 2 - Number of commits pushed per repository
Step3:  The Split-Apply-Combine Pattern
Hadley Wickham <img src="http
Step4:  <img src="http
Step5:  This
filters out only the "PushEvent"s
splits the dataset by repository
sums the commits for each group
combines the groups and their sums into a dictionary
Pandas - Python Data Analysis Library
<p><a href="http
Step6:  Example 1 (using Pandas) - Number of Repositories Created
Step7:  Example 2 (using Pandas) - Number of commits pushed per repo
Step8:  Example 1 - revisited
Step9:  Great for interactive work
Step10:  Would prefer to write
events | map(...) | reduce(...)
Example 1 - pipelined
Step11:  PyToolz
Example 1 - pipeline using PyToolz
Step12:  Example 2 - pipelined with PyToolz
Step13:  The Point of Learning Patterns
From Cosma Shalizi's Statistical Computing course
Step14:  New tools
Blaze
<img src="img/blaze_med.png" width=400>
Dask
<img src="img/dask-collections-schedulers.png">
Example 2 - using blaze (and pandas)
Step15:  You can run the same computation on different backends!
Step16:  Dask and Castra
<ASSISTANT_TASK:>
Python Code:
import os
import gzip
import ujson as json
directory = 'data/github_archive'
filename = '2015-01-29-16.json.gz'
path = os.path.join(directory, filename)
with gzip.open(path) as f:
        events = [json.loads(line) for line in f]
#print json.dumps(events[0], indent=4)
Explanation: The Split-Apply-Combine Pattern in Data Science and Python
Tobias Brandt
<img src="img/argon_logo.png" align=left width=200>
<!-- <img src="http://www.argonassetmanagement.co.za/css/img/logo.png" align=left width=200> -->
<img src='img/argon_website.png' align='middle'>
Google trends chart
Data Science
According to https://en.wikipedia.org/wiki/Data_science:
In November 1997, C.F. Jeff Wu gave the inaugural lecture entitled "Statistics = Data Science?"[5] for his appointment to the H. C. Carver Professorship at the University of Michigan.[6] In this lecture, he characterized statistical work as a trilogy of data collection, data modeling and analysis, and decision making. In his conclusion, he initiated the modern, non-computer science, usage of the term "data science" and advocated that statistics be renamed data science and statisticians data scientists.[5]
The Github Archive Dataset
https://www.githubarchive.org/
Open-source developers all over the world are working on millions of projects: writing code & documentation, fixing & submitting bugs, and so forth. GitHub Archive is a project to record the public GitHub timeline, archive it, and make it easily accessible for further analysis.
GitHub provides 20+ event types, which range from new commits and fork events, to opening new tickets, commenting, and adding members to a project. These events are aggregated into hourly archives, which you can access with any HTTP client:
gzipped json files
yyyy-mm-dd-HH.json.gz
End of explanation
new_repo_count = 0
for event in events:
    new_repo_count += \
        1 if event['type']=="CreateEvent" else 0
print new_repo_count
Explanation: <pre>
{
    <b>"payload": {</b>
        "master_branch": "master", 
        "ref_type": "branch", 
        "ref": "disable_dropdown", 
        "description": "OOI UI Source Code", 
        "pusher_type": "user"
    }, 
    <b>"created_at": "2015-01-29T16:00:00Z", </b>
    "actor": {
        "url": "https://api.github.com/users/birdage", 
        <b>"login": "birdage", </b>
        "avatar_url": "https://avatars.githubusercontent.com/u/547228?", 
        "id": 547228, 
        "gravatar_id": ""
    }, 
    "id": "2545235518", 
    "repo": {
        "url": "https://api.github.com/repos/birdage/ooi-ui", 
        "id": 23796192, 
        <b>"name": "birdage/ooi-ui"</b>
    }, 
    <b>"type": "CreateEvent", </b>
    "public": true
}
</pre>
Typical Questions
How many Github repositories are created per hour/day/month?
To which repositories are the most commits are pushed per hour/day/month?
Which projects receive the most pull requests?
What are the most popular languages on Github?
Example 1 - Number of Repositories Created
End of explanation
repo_commits = {}
for event in events:
    if event['type']=="PushEvent":
        repo = event['repo']['name']
        commits = event['payload']['size']
        repo_commits[repo] = \
            repo_commits.get(repo, 0) + commits 
def print_top_items(dct, N=5):
    sorted_items = sorted(
        dct.iteritems(), key=lambda t: t[1], reverse=True)
    for key, value in sorted_items[:N]:
        print "{:40} {}".format(key, value)
print_top_items(repo_commits)
Explanation: Example 2 - Number of commits pushed per repository
End of explanation
from IPython.display import HTML
HTML('<iframe src="http://www.jstatsoft.org/v40/i01" width=800 height=400></iframe>')
Explanation: The Split-Apply-Combine Pattern
Hadley Wickham <img src="http://pix-media.s3.amazonaws.com/blog/1001/HadleyObama.png" width=250 align=left>
Hadley Wickham, the man who revolutionized R
*If you don’t spend much of your time coding in the open-source statistical programming language R, 
his name is likely not familiar to you -- but the statistician Hadley Wickham is, 
in his own words, “nerd famous.” The kind of famous where people at statistics conferences 
line up for selfies, ask him for autographs, and are generally in awe of him.
End of explanation
repo_commits = {}
for event in events:
    if event['type']=="PushEvent":
        repo = event['repo']['name']
        commits = event['payload']['size']
        repo_commits[repo] = \
            repo_commits.get(repo, 0) + commits 
print_top_items(repo_commits)
Explanation: <img src="http://i.imgur.com/CoJHnAF.jpg">
StackOverflow: split-apply-combine tag
Pandas documentation: Group By: split-apply-combine
PyTools documentation: Split-apply-combine with groupby and reduceby
Blaze documentation: Split-Apply-Combine - Grouping
R plyr: plyr: Tools for Splitting, Applying and Combining Data
Julia documentation: The Split-Apply-Combine Strategy
The Basic Pattern
Split the data by some grouping variable
Apply some function to each group independently
Combine the data into some output dataset
The apply step is usually one of
aggregate
transform
or filter
Example 2 - examined
End of explanation
import numpy as np
import pandas as pd
from collections import namedtuple
GithubEvent = namedtuple('GithubEvent', ['type_', 'user', 'repo', 'created_at', 'commits'])
def make_record(event):
    return GithubEvent(
        event['type'], event['actor']['login'], 
        event['repo']['name'], pd.Timestamp(event['created_at']),
        event['payload']['size'] if event['type']=='PushEvent' else np.nan
        )
df = pd.DataFrame.from_records(
    (make_record(ev) for ev in events),
    columns=GithubEvent._fields, index='created_at')
df.head()
Explanation: This
filters out only the "PushEvent"s
splits the dataset by repository
sums the commits for each group
combines the groups and their sums into a dictionary
Pandas - Python Data Analysis Library
<p><a href="http://pandas.pydata.org/"><img src="http://pandas.pydata.org/_static/pandas_logo.png" align=right width=400></a></p>
Provides high-performance, easy-to-use data structures and data analysis tools.
Provides core data structure DataFrame
pandas.DataFrame
Basically in-memory database tables (or spreadsheets!)
Tabular data that allows for columns of different dtypes
Labeled rows and columns (index)
Hierarchical indexing allows for representing Panel data
End of explanation
df[df.type_=='CreateEvent'].head()
len(df[df.type_=='CreateEvent'])
Explanation: Example 1 (using Pandas) - Number of Repositories Created
End of explanation
repo_commits = {}
for event in events:
    if event['type']=="PushEvent":
        repo = event['repo']['name']
        commits = event['payload']['size']
        repo_commits[repo] = \
            repo_commits.get(repo, 0) + commits 
print_top_items(repo_commits)
repo_commits = df[df.type_=='PushEvent'].groupby('repo').commits.sum()
repo_commits.sort(ascending=False)
repo_commits.head(5)
Explanation: Example 2 (using Pandas) - Number of commits pushed per repo
End of explanation
event_counts = df.groupby('type_').repo.count()
event_counts.sort(ascending=False)
event_counts.head()
Explanation: Example 1 - revisited
End of explanation
new_repo_count = 0
for event in events:
    new_repo_count += \
        1 if event['type']=="CreateEvent" else 0
print new_repo_count
reduce(lambda x,y: x+y, 
       map(lambda ev: 1 if ev['type']=='CreateEvent' else 0, 
           events))
Explanation: Great for interactive work:
tab-completion!
inspect data with df.head() & df.tail()
quick overview of data ranges with df.describe()
<img src="http://i.imgur.com/6b2AF7e.jpg" width=300 align='middle'>
However ...
Pandas currently only handles in-memory datasets!
So not suitable for big data!
<img src="img/devops_borat.jpg">
MapReduce
"If you want to process Big Data, you need some MapReduce framework like one of the following"
<p>
<a href="https://hadoop.apache.org/"><img src="https://hadoop.apache.org/images/hadoop-logo.jpg" width=200 align=left></a>
<a href="http://spark.apache.org/"><img src="http://spark.apache.org/images/spark-logo.png" width=100 align=left></a>
</p>
<img src="https://mitpress.mit.edu/sicp/full-text/book/cover.jpg" align=right width=150>
The key to these frameworks is adopting a functional [programming] mindset. In Python this means, think iterators!
See The Structure and Interpretation of Computer Programs
(the "Wizard book")
in particular Chapter 2 Building Abstractions with Data 
and Section 2.2.3 Sequences as Conventional Interfaces
Luckily, the Split-Apply-Combine pattern is well suited to this!  
Example 1 - revisited
End of explanation
def datapipe(data, *transforms):
    for transform in transforms:
        data = transform(data)
    return data
datapipe(
    events,
    lambda events: map(lambda ev: 1 if ev['type']=='CreateEvent' else 0, events),
    lambda counts: reduce(lambda x,y: x+y, counts)
    )
Explanation: Would prefer to write
events | map(...) | reduce(...)
Example 1 - pipelined
End of explanation
from toolz.curried import pipe, map, reduce
pipe(events,
     map(lambda ev: 1 if ev['type']=='CreateEvent' else 0),
     reduce(lambda x,y: x+y)
    )
Explanation: PyToolz
Example 1 - pipeline using PyToolz
End of explanation
repo_commits = {}
for event in events:
    if event['type']=="PushEvent":
        repo = event['repo']['name']
        commits = event['payload']['size']
        repo_commits[repo] = \
            repo_commits.get(repo, 0) + commits 
print_top_items(repo_commits)
from toolz.curried import filter, reduceby
pipe(events,
     filter(lambda ev: ev['type']=='PushEvent'),
     reduceby(lambda ev: ev['repo']['name'],
              lambda commits, ev: commits+ev['payload']['size'],
              init=0),
     print_top_items
     )
Explanation: Example 2 - pipelined with PyToolz
End of explanation
def count_commits(filename):
    import gzip
    import json
    from toolz.curried import pipe, filter, reduceby
    with gzip.open(filename) as f:
        repo_commits = pipe(
            map(json.loads, f),
            filter(lambda ev: ev['type']=='PushEvent'),
            reduceby(lambda ev: ev['repo']['name'],
                     lambda commits, e: commits+e['payload']['size'],
                     init=0)
                     )
    return repo_commits
print_top_items(count_commits(path))
import glob
files = glob.glob('C:/ARGO/talks/split-apply-combine/data/github_archive/2015-01-*')
print len(files)
N = 24  #len(files)    # 10
%%time
from toolz.curried import reduceby
from __builtin__ import map as pmap
repo_commits = \
    pipe(pmap(count_commits, files[:N]),
         lambda lst: reduce(lambda out, dct: out + dct.items(), lst, []),
         reduceby(lambda t: t[0], lambda s,t: s+t[1], init=0)
         )
print_top_items(repo_commits)
%%time
# Remember to start the ipcluster!
# ipcluster start -n 4
from IPython.parallel import Client
p = Client()[:]
pmap = p.map_sync
repo_commits = \
    pipe(pmap(count_commits, files[:N]),
         lambda lst: reduce(lambda out, dct: out + dct.items(), lst, []),
         reduceby(lambda t: t[0], lambda s,t: s+t[1], init=0)
         )
print_top_items(repo_commits)
Explanation: The Point of Learning Patterns
From Cosma Shalizi's Statistical Computing course:
Distinguish between what you want to do and how you want to do it.
Focusing on what brings clarity to intentions.
How also matters, but can obscure the high level problem.
Learn the pattern, recognize the pattern, love the pattern!
Re-use good solutions!
Iteration Considered Unhelpful
Could always do the same thing with for loops, but those are
verbose - lots of "how" obscures the "what"
painful/error-prone bookkeeping (indices, placeholders, ...)
clumsy - hard to parallelize
Out-of-core processing - toolz example
End of explanation
repo_commits = df[df.type_=='PushEvent'].groupby('repo').commits.sum()
repo_commits.sort(ascending=False)
repo_commits.head(5)
from blaze import Symbol, by
event = Symbol('event', 'var * {created_at: datetime, type_: string, user: string, repo: string, commits: int}')
push_events = event[event.type_=='PushEvent']
repo_commits = by(push_events.repo, commits=push_events.commits.sum())
top_repos = repo_commits.sort('commits', ascending=False).head(5)
from blaze import compute
print compute(top_repos, df)
Explanation: New tools
Blaze
<img src="img/blaze_med.png" width=400>
Dask
<img src="img/dask-collections-schedulers.png">
Example 2 - using blaze (and pandas)
End of explanation
from odo import odo
uri = 'sqlite:///data/github_archive.sqlite::event'
odo(df, uri)
from blaze import Data
db = Data(uri)
compute(top_repos, db)
import os
if os.path.exists('data/github_archive.sqlite'):
    os.remove('data/github_archive.sqlite')
Explanation: You can run the same computation on different backends!
End of explanation
from castra import Castra
castra = Castra('data/github_archive.castra',
                template=df, categories=categories)
castra.extend_sequence(map(to_df, files), freq='1h')
import dask.dataframe as dd
from dask.diagnostics import ProgressBar
pbar = ProgressBar()
pbar.register()
df = dd.from_castra('data/github_archive.castra')
df.head()
df.type.value_counts().nlargest(5).compute()
df[df.type=='PushEvent'].groupby('repo').commits.resample('h', how='count').compute()
Explanation: Dask and Castra
End of explanation
<END_TASK>