###
# There is now a 'SparkContext' instance available as the named variable 'sc'
# and there is a HiveContext instance (for SQL-like queries) available as 'sqlCtx'
#
## Check that this simple code runs without error:
sc.parallelize([1,2,3,4,5]).take(2)

###
# Inspect the SparkContext [sc] or the HiveContext [sqlCtx]
#help(sc)
help(sqlCtx)

"""
Explanation: Work work, money money
Combine job board and BLS data to find trends in job / industry growth in Chicago and elsewhere.
Data:

BLS: http://www.bls.gov/help/hlpforma.htm,
Occupational Employment Statistics http://download.bls.gov/pub/time.series/oe/
National Compensation Survey http://download.bls.gov/pub/time.series/nc/


Census:
county shapefiles https://www.census.gov/geo/maps-data/data/tiger-cart-boundary.html
or, better, Mike Bostock's US-atlas project (census shapefiles ⇒ GeoJSON)


JobsAggregator: (results from all 5 of Indeed, SimplyHired, CareerBuilder, Monster, and CareerJet): 

Technology:

Apache Spark on an Amazon EC2 (Elastic Cloud 2) cluster; instructions below
input data stored in Amazon S3 buckets, output written to HDFS permanent storage
Images rendered in D3 via a private Lightning server

Steps to launch Spark AWS EC2 cluster
Here is how to do your own:

Get an Amazon Web Services account

Get the newest version of Spark, pre-built for Hadoop 2.4. It has to be
   pre-built so that the pyspark client will have the proper jars
curl -O http://www.apache.org/dyn/closer.cgi/spark/spark-1.3.0/spark-1.3.0-bin-hadoop2.4.tgz
tar -xzvf spark-1.3.0-bin-hadoop2.4.tgz



Launch it on Amazon EC2 using a script in the ec2 directory in the source instructions 
cd spark-1.3.0-bin-hadoop2.4
export SPARK_HOME=`pwd`
$SPARK_HOME/ec2/spark-ec2  --slaves 2  \
    --key-pair &lt;Amazon_Keypair_Name&gt;  \
    --identity-file &lt;path/to/Amazon_Keypair.pem&gt;  \
    --copy-aws-credentials  \
    --zone us-east-1b --instance-type=m1.medium  \
    launch spark_cluster



At the end of the startup run, it will show a URL we can use to connect, or else
   navigate to the EC2 dashboard through the Amazon Web Service Console
   to find out what the IP address is for the master node. The Spark dashboard is on
   port 8080 by default:  &lt;ip address&gt;:8080 <br/>
   Mine is here: http://ec2-54-166-72-95.compute-1.amazonaws.com:8080/


Steps to launch IPython notebook connecting to AWS EC2 cluster
A handful of blogs describe how to set up IPython + Spark; they're helpful but outdated:

Presentation of two distinct ways to do it
Too much information
Two separate ways presented as if they were one

IPython options are built into pyspark. Follow option 3 above to create a password-protected IPython notebook configuration file, but instead of setting c.NotebookApp.ip = * like it says, use the Master's designated address. (e.g. 'ec2-54-166-72-95.compute-1.amazonaws.com').
Configuration for Python 2.7
PySpark prefers Python 2.7 but Python 2.6 is the default Python for an Amazon EC2 instance. The below are a set of scripts to install Python 2.7 and some dependencies I need (requests, pymongo and lxml). All of the others are PySpark dependencies.
SSH in to the master node and execute:
xargs -L1 -a commands.txt ./go.sh

with commands.txt containing:
yes | yum install python27-devel
unlink /etc/alternatives/python
ln -s /usr/bin/python2.7 /etc/alternatives/python

wget https://bootstrap.pypa.io/ez_setup.py
python ez_setup.py
easy_install pip
rm ez_setup.py
rm setuptools-16.0.zip

easy_install Cython
yes | yum install freetype-devel
yes | yum install libpng-devel
pip install numpy scipy
pip install matplotlib

yes | yum install libxml2-devel
yes | yum install libxslt-devel
pip install requests lxml pymongo
pip install ipython[notebook]

and go.sh containing:
#!/usr/bin/env bash

echo '------------------------------------------------'
echo $@;  $@

while read worker
do
  ssh ${worker} "echo 'machine ${worker}'; $@"
done &lt; /root/spark/conf/slaves

Back to launching an IPython notebook


Add a Custom TCP rule to the spark_cluster-master Amazon EC2 security group to allow the port for the IPython notebook (8888 by default; Cloudera says to watch for potential port assignment collisions but 8888 worked fine.)
    a. Navigate  to the EC2 console
    b. Click on the master instance, and then on the security group assigned to that instance
    c. It will open another user interface. Click on Actions → Edit inbound rules and add a custom TCP rule with protocol TCP, port range 8888, and source Anywhere


Set environment variables for &lt;spark-home&gt;/bin/pyspark to use to launch a properly configured IPython notebook, ready to use pyspark. As of Spark 1.2:
export PYSPARK_DRIVER_PYTHON=ipython
export PYSPARK_DRIVER_PYTHON_OPTS='notebook --profile=pyspark'



Launch pyspark. Specifically designate the master node or else pyspark will run as a local standalone spark instance:
/root/spark/bin/pyspark --master spark://ec2-54-166-72-95.compute-1.amazonaws.com:7077



And to ensure persistence after logging out, the above was wrapped in nohup &lt;command&gt; &amp;
End of explanation
"""


from random import random
from operator import add

def monte_carlo(_):
    """4 * area (1 quadrant of a unit circle)  pi"""
    x = random()
    y = random()
    return 4.0 if pow(x, 2) + pow(y, 2) < 1 else 0

N = 1000
parts = 2
sc.parallelize(xrange(N), parts).map(monte_carlo).reduce(add) / N

"""
Explanation: De Pie :: (parallel calculation)
${SPARK_HOME}/spark/examples/src/main/python/pi.py

<img src="http://www.mixingbowlgal.com/wp-content/uploads/2013/06/IMG_3999.jpg" style="width:250px;"></img>
End of explanation
"""


### ------------------------------------------------- AMAZON ----- ###
# ⇒ These files identify columns that will be common to the job
#    board data and the BLS datasets.
#
# To use S3 buckets add `--copy-aws-credentials` to the ec2 launch command.
#
# Create a Resilient Distributed Dataset with the
# list of occupations in the BLS dataset:
#     https://s3.amazonaws.com/tts-wwmm/occupations.txt
from pyspark.sql import Row

# Load the occupations lookups and convert each line to a Row.
lines = sc.textFile('s3n://tts-wwmm/occupations.txt')
Occupation = Row('OCC_CODE', 'OCC_TITLE')
occ = lines.map(lambda l: Occupation( *l.split('\t') ))

# Do the same for the areas lookups.
lines = sc.textFile('s3n://tts-wwmm/areas.txt')
Area = Row('AREA', 'AREA_NAME')
area = lines.map(lambda l: Area( *l.split('\t') ))
area_df = sqlCtx.createDataFrame(area)
area_df.registerTempTable('area')

# Just to show how sqlCtx.sql works
states = sqlCtx.sql("SELECT AREA_NAME, AREA FROM area WHERE AREA RLIKE '^S.*'")
print states.take(2)

# Same as above, but result is another Resilient Distributed Dataset
states = area.filter(lambda a: a.AREA.startswith('S'))

# Create every combination of occupation, state
occ_by_states = occ.cartesian(states)

# Broadcast makes a static copy of the variable available to all nodes
#broadcast_state_names = sc.broadcast(broadcast_state_names)
#
#print broadcast_state_names.take(2)


"""
Explanation: The data
We will merge a job postings dataset and the BLS Occupations and Earnings data together, using location, occupation, and possibly industry category.
Amazon S3 buckets
Information on connecting to S3 from Spark is at the bottom of the
Spark docs on EC2 scripts, and copied here:

You can specify a path in S3 as input through a URI of the form s3n://&lt;bucket&gt;/path. To provide AWS credentials for S3 access, launch the Spark cluster with the option --copy-aws-credentials. Full instructions on S3 access using the Hadoop input libraries can be found on the Hadoop S3 page.
In addition to using a single input file, you can also use a directory of files as input by simply giving the path to the directory.

Separate files listing the BLS categories were loaded to an Amazon S3 bucket: tts-wwmm/areas.txt, tts-wwmm/industry.txt, tts-wwmm/occupations.txt. All are two-column files with no headers, and a tab separating the variable code and the variable label.
The JobsAggregator data
JobsAggregator aggregates from Indeed, SimplyHired, CareerBuilder, Monster, and CareerJet, showing the most recent job posts on each site.
The function scrape in the file jobs_aggregator_scraper.py iteratively scrapes the site, and returns a generator that yields current job listings (as a dictionary) for a given state and occupation.
The BLS data
The data are actually not best pulled via the BLS API. It has only the most recent year's statistics; the rest are archived at http://www.bls.gov/oes/tables.htm.
The contents of the archive files were loaded to a MongoDB database. There was manual work to handle different column names and file formats for the different years. Data are available by occupation at the state and national level, and at more aggregated levels for municipal areas. Below is an example of one observation. The ANNUAL and OVERALL entries are lists of dictionaries with one entry per year, possibly with data from as far back as 2000.
{    "AREA": "3800003",
"AREA_NAME": "East Central North Dakota",
       "ST": "ND",
 "OCC_CODE": "39-1012",
"OCC_TITLE": "Slot key persons",
   "ANNUAL": [
        {
             "YEAR": 2009,
            "pct90": 34530,
            "pct75": 32730,
            "pct50": 30170,
            "pct25": 27610,
            "pct10": 19250
        }
    ],
    "OVERALL": [
        {
               "YEAR": 2009,
          "JOBS_1000": 0.923,
            "TOT_EMP": 40,
             "A_MEAN": 29100,
          "MEAN_PRSE": 5.6,
             "H_MEAN": 13.99,
           "EMP_PRSE": 30.2
        }
    ]
}
Load the lookup tables from Amazon S3
End of explanation
"""


### ----------------------------------------- JOBS_AGGREGATOR ----- ###
#
# Make `jobs_aggregator_scraper.py` available on all nodes
# and iteratively get the top 5 jobs from each poster in each state for
# each occupation via JobsAggregator.com
sc.addPyFile('s3n://tts-wwmm/jobsaggregator_scraper.py')

def scrape_occupation(occ_state):
    from jobsaggregator_scraper import scrape
    occ_row, state_entry = occ_state
    return [Row(**job)
            for job in scrape(state=state_entry[1], occupation=occ_row.OCC_TITLE)]

jobs = occ_by_states.flatMap(scrape_occupation).distinct()
jobs_df = sqlCtx.inferSchema(jobs)
jobs_df.registerTempTable('jobs')

jobs_df.toJSON().saveAsTextFile('wwmm/jobsaggregator_json')

jobs.saveAsTextFile('wwmm/jobsaggregator_df')

jobs.take(2)

"""
Explanation: Scrape JobsAggregator
End of explanation
"""


### -------------------------------------------- BLS OES DATA ----- ###
#
# The OES data were loaded to a mongolabs database. Read the URI
# (which has a user name and password) from an environment variable
# and create a connection. The pymongo API is very simple.
#
# Datasets are stored one entry per Occupation ID (OCC_ID)
# per area (00-0000)
from pymongo import MongoClient

MONGO_URI = os.getenv('MONGO_URI')
client = MongoClient(MONGO_URI)  # connection
oe = client.oe  # database

# Confirm we can get data from each collection
oo = oe['nat'].find(filter={'OCC_CODE':'00-0000'},
               projection={'_id':False,
                           'OCC_CODE':True, 'OCC_TITLE':True,
                           'ANNUAL':{'$slice':-5}, 'OVERALL':{'$slice':-2}})

for o in oo:
    print o

# Which OCC contains software-type people?
occ_df = sqlCtx.createDataFrame(occ)
occ_df.registerTempTable('occ')

computer_jobs = sqlCtx.sql((
    "SELECT OCC_CODE, OCC_TITLE "
     "FROM occ "
     "WHERE OCC_TITLE RLIKE 'omputer'"
    )).collect()

for row in computer_jobs:
    print "{OCC_CODE}: {OCC_TITLE}".format(**row.asDict())

# Want Chicago's area code

chicago = sqlCtx.sql((
    "SELECT AREA, AREA_NAME "
     "FROM area "
     "WHERE AREA_NAME RLIKE 'icago' or AREA_NAME RLIKE 'llinois'"
    )).collect()

print "\n".join("{}: {}".format(c.AREA, c.AREA_NAME) for c in chicago)

# Now get the data:
## -------------------------------------- National

desired_data = {'_id':False,
                'ANNUAL':{'$slice':-5}, 'OVERALL':{'$slice':-2}}
nat = oe['nat'].find(filter={'OCC_CODE':'15-1131'},
               projection=desired_data)

nat = [n for n in nat]
len(nat)

## -------------------------------------- State

il = oe['st'].find(filter={'OCC_CODE':'15-1131',
                           'AREA':'17'},
                  projection=desired_data)

il = [i for i in il]
len(il)

## -------------------------------------- Municipal Areas
## The lookup for chicago didn't work...
## ... so I am looking through all of the municipal areas...
chi = oe['ma'].find(filter={'OCC_CODE':'15-1131'},
                   projection=desired_data)
chi = [c for c in chi if 'IL' in c['AREA_NAME']]
len(chi)

# Get the mean 
import tablib

nat_annual = tablib.Dataset()
nat_annual.dict = nat[0]['ANNUAL']
il_annual = tablib.Dataset()
il_annual.dict = il[0]['ANNUAL']
chi_annual = tablib.Dataset()
chi_annual.dict = chi[1]['ANNUAL']

"""
Explanation: Load OES Data
End of explanation
"""


from lightning import Lightning

lgn = Lightning(host="https://tts-lightning.herokuapp.com",
                ipython=True,
                auth=("tanya@tickel.net", "password"))

# Median salaries
lgn.line(series=[nat_annual['pct50'], il_annual['pct50'], chi_annual['pct50']],
         index=nat_annual['YEAR'],
         color=[[0,0,0],[255,0,0],[0,155,0]],
         size=[5,2,2],
         xaxis="Year",
         yaxis="Median annual salary")

# How about regionally?

all_states = oe['st'].find(filter={'OCC_CODE':'15-1131'},
                  projection={'$_id': False,
                             'OVERALL':{'$slice':-2}})
all_states = [a for a in all_states]
len(all_states)

state_abbrs = [a['ST'] for a  in all_states]
mean_salaries = [a['OVERALL'][0]['A_MEAN'] for a in all_states]
num_employed = [a['OVERALL'][0]['TOT_EMP'] for a in all_states]

# Mean salaries
print "max average salary:", max(mean_salaries)
print "Illinois:", mean_salaries[state_abbrs.index('IL')]
lgn.map(regions=state_abbrs, values=mean_salaries)

# Employees
print "Most programmers:", max(num_employed)
print "Illinois:", num_employed[state_abbrs.index('IL')]
lgn.map(regions=state_abbrs, values=num_employed)

salaries = tablib.Dataset(*zip(state_abbrs, mean_salaries),
                          headers=('State', 'Salary'))
employees= tablib.Dataset(*zip(state_abbrs, num_employed),
                         headers=('State', 'Employees'))

salaries = salaries.sort("Salary", reverse=True)
print "\n".join("{s[0]}: {s[1]:0,.0f}".format(s=s) for s in salaries[:5])

employees = employees.sort("Employees", reverse=True)
print "\n".join("{e[0]}: {e[1]:0,.0f}".format(e=e) for e in employees[:5])

"""
Explanation: Lightning-viz plots for inline D3.js in IPython
http://lightning-viz.org/
End of explanation
"""


