Hadoop Experiment - MapReduce on Cloudera

In this example I will extract data with NES reviews from http://videogamecritic.com. I will create a dataframe, add some extra fields and save the data to a CSV-file. This file will be used for a simple MapReduce script.

Note: I have a Docker container running with Selenium instead of installing all dependencies on my system. See this page.

Extract script

The script below is used to retrieve the data. It is ugly code, but it is doing the job.

import lxml.html
import requests
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
import pandas as pd

URL = 'http://videogamecritic.com/nes.htm'

def map_rating(rating):
    """ Function to convert the rating to a number """
    if (rating == "A+"): return 1;
    if (rating == "A"): return 2;
    if (rating == "A-"): return 3;
    if (rating == "B+"): return 4;
    if (rating == "B"): return 5;
    if (rating == "B-"): return 6;
    if (rating == "C+"): return 7;
    if (rating == "C"): return 8;
    if (rating == "C-"): return 9;
    if (rating == "D+"): return 10;
    if (rating == "D"): return 11;
    if (rating == "D-"): return 12;
    if (rating == "F"): return 13;
    if (rating == "F-"): return 14;
    return 15;

resp = requests.get(URL)
if resp.status_code != 200:
    raise Exception('GET '+link+' {}'.format(resp.status_code))
tree= lxml.html.fromstring(resp.content)
gamepages = ['http://videogamecritic.com/' + game.get('href') for game in tree.cssselect('h3 a')]
driver = webdriver.Remote("http://localhost:4444/wd/hub", DesiredCapabilities.CHROME)
gamedata = []
for page in gamepages:
    # Retrieve the data
    driver.get(page)
    data = lxml.html.fromstring(driver.page_source)
    # Extract the fields
    grades = [elem.text_content() for elem in data.cssselect('div[class*=\'hdr\']')]
    names = [elem.text_content() for elem in data.cssselect('div[class*=\'hdl\']')]
    metadata = [elem.text_content() for elem in data.cssselect('div[class*=\'mdl\']')]
    votes = [elem.text_content() for elem in data.cssselect('div[class*=\'vote\']')]
    # Append to dataset
    gamedata += list(zip(names, votes, grades, metadata))
driver.quit()
##
# DataFrame magic
##
df = pd.DataFrame.from_dict(gamedata)
df = df.rename(columns={0: "name", 1: "vote", 2: "grade", 3: "publisher"})
# Extract and convert data
df['reader_rating'] = df['vote'].str.extract('Readers:\s(.*?)\s\(', expand=True)
df['reader_rating']  = df['reader_rating'].apply(lambda x: map_rating(x)).astype('int')
df['number_of_votes'] = df['vote'].str.extract('\((\d*)\svotes\)', expand=True).astype('int')
df['grade'] = df['grade'].str.replace("Grade:", "").str.strip().apply(lambda x: map_rating(x)).astype('int')
df['publish_year'] = df['publisher'].str.extract('\((\d*)\)Reviewed', expand=True)
df['publisher'] = df['publisher'].str.extract("Publisher:\s(.*?)\s\(", expand=True)
df.drop('vote', axis=1, inplace=True)
# Calculate the total grade
df['total_grade'] = (df['grade'] + df['reader_rating']*df['number_of_votes']) / (df['number_of_votes']+1)
# Corrections
df['publisher'] = df['publisher'].str.replace('Electrobrain', 'Electro Brain')
# Save to file
df.to_csv('nesgamedata.csv', sep='\t', header=False)

This will result in the following file.

0   10-Yard Fight   12  Nintendo    10  44  1985    10.044444444444444
1   1942    11  Capcom  8   65  1985    8.045454545454545
2   1943    5   Capcom  4   58  1988    4.016949152542373
3   720 Degrees 13  Tengen  11  24  1989    11.08
4   8 Eyes  13  Taxan   10  30  1989    10.096774193548388
5   Abadox  11  Abadox  6   34  1989    6.142857142857143
6   Adventure Island    7   Hudson Soft 6   63  1987    6.015625
7   Adventure Island 2  4   Hudson Soft 5   40  1990    4.975609756097561
8   Adventure Island 3  4   Hudson Soft 5   27  1992    4.964285714285714
9   Adventures in the Magic Kingdom 5   Capcom  8   23  1990    7.875
...

MapReduce script

Calculate how often what rating is used.

  1. Map the reader ratings
  2. Reduce to counts per rating

Lets create the script that will perform my first MapReduce action. The content of the following cell will be saved to a file which can in turn be used to perform the mapping and reducing.

from mrjob.job import MRJob
from mrjob.step import MRStep

class GamesBreakdown(MRJob):
    def steps(self):
        return [
            MRStep(mapper=self.mapper_get_ratings,
                   reducer=self.reducer_count_ratings)
        ]

    def mapper_get_ratings(self, _, line):
        (index, name, grade, publisher, reader_rating, number_of_votes, publish_year, total_grade) = line.split('\t') 
        yield reader_rating, 1

    def reducer_count_ratings(self, key, values):
        yield key, sum(values)

if __name__ == '__main__':
    GamesBreakdown.run()

Execution

To use Hadoop, the command should look like the following and should be run on the Hadoop machine, with the hadoop-streaming-jar argument only given in case the .jar is not found:

python gamesbreakdown.py -r hadoop --hadoop-streaming-jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar nesgamedata.csv

To do this, upload the nesgamesdata.csv file and the script gamesbreakdown.py to HDFS, using the Hue files view and upload functionality. Next copy them to the local folder to be used in the command. Of course other methods to get the files locally on the Hadoop machine can be used.

I am using the Cloudera Quickstart image to experiment with Docker and Cloudera and the docker-compose.yml contains the following:

version: '2'
services:
  cloudera:
    hostname: quickstart.cloudera
    command: /usr/bin/docker-quickstart
    tty: true
    privileged: true
    image: cloudera/quickstart:latest
    volumes:
      - ./data:/opt/data
    ports:
      - "8020:8020"
      - "8022:22"     # ssh
      - "7180:7180"   # Cloudera Manager
      - "8888:8888"   # HUE
      - "11000:11000" # Oozie
      - "50070:50070" # HDFS REST Namenode
      - "2181:2181"
      - "11443:11443"
      - "9090:9090"
      - "8088:8088"
      - "19888:19888"
      - "9092:9092"
      - "8983:8983"
      - "16000:16000"
      - "16001:16001"
      - "42222:22"
      - "8042:8042"
      - "60010:60010"
      - "8080:8080"
      - "7077:7077"

Connect to the machine and verify the file is present:

[email protected]:~/cloudera_docker$ docker exec -ti clouderadocker_cloudera_1 bash
[[email protected] /]# cd home/
[[email protected] home]# mkdir gamedata && cd $_
[[email protected] gamedata]# hadoop fs -get gamedata/gamesbreakdown.py gamesbreakdown.py
[[email protected] gamedata]# hadoop fs -get gamedata/nesgamedata.csv nesgamedata.csv
[[email protected] gamedata]# ll
total 20
-rw-r--r-- 1 root root   590 Sep  5 13:42 gamesbreakdown.py
-rw-r--r-- 1 root root 15095 Sep  5 13:56 nesgamedata.csv

Install the mrjob library on the Cloudera container.

[[email protected] gamedata]# yum install python-pip -y
[[email protected] gamedata]# pip install mrjob

Run the script without using Hadoop to verify the installation.

[[email protected] gamedata]# python gamesbreakdown.py nesgamedata.csv
No configs found; falling back on auto-configuration
Creating temp directory /tmp/gamesbreakdown.root.20170905.135809.859796
Running step 1 of 1...
Streaming final output from /tmp/gamesbreakdown.root.20170905.135809.859796/output...
"10"    15
"11"    14
"12"    12
"13"    4
"2"     9
"3"     28
"4"     44
"5"     34
"6"     48
"7"     31
"8"     28
"9"     15
Removing temp directory /tmp/gamesbreakdown.root.20170905.135809.859796...

Now use Hadoop to start the cluster magic.

[[email protected] gamedata]# python gamesbreakdown.py -r hadoop nesgamedata.csv
No configs found; falling back on auto-configuration
Looking for hadoop binary in $PATH...
Found hadoop binary: /usr/bin/hadoop
Using Hadoop version 2.6.0
Looking for Hadoop streaming jar in /home/hadoop/contrib...
Looking for Hadoop streaming jar in /usr/lib/hadoop-mapreduce...
Found Hadoop streaming jar: /usr/lib/hadoop-mapreduce/hadoop-streaming.jar
Creating temp directory /tmp/gamesbreakdown.root.20170905.135908.241472
Copying local files to hdfs:///user/root/tmp/mrjob/gamesbreakdown.root.20170905.135908.241472/files/...
Running step 1 of 1...
  packageJobJar: [] [/usr/jars/hadoop-streaming-2.6.0-cdh5.7.0.jar] /tmp/streamjob3556720292725669631.jar tmpDir=null
  Connecting to ResourceManager at /0.0.0.0:8032
  Connecting to ResourceManager at /0.0.0.0:8032
  Total input paths to process : 1
  number of splits:2
  Submitting tokens for job: job_1504618114531_0001
  Submitted application application_1504618114531_0001
  The url to track the job: http://quickstart.cloudera:8088/proxy/application_1504618114531_0001/
  Running job: job_1504618114531_0001
  Job job_1504618114531_0001 running in uber mode : false
   map 0% reduce 0%
   map 50% reduce 0%
   map 100% reduce 0%
   map 100% reduce 100%
  Job job_1504618114531_0001 completed successfully
  Output directory: hdfs:///user/root/tmp/mrjob/gamesbreakdown.root.20170905.135908.241472/output
Counters: 49
        File Input Format Counters
                Bytes Read=19191
        File Output Format Counters
                Bytes Written=86
        File System Counters
                FILE: Number of bytes read=2307
                FILE: Number of bytes written=358424
                FILE: Number of large read operations=0
                FILE: Number of read operations=0
                FILE: Number of write operations=0
                HDFS: Number of bytes read=19527
                HDFS: Number of bytes written=86
                HDFS: Number of large read operations=0
                HDFS: Number of read operations=9
                HDFS: Number of write operations=2
        Job Counters
                Data-local map tasks=2
                Launched map tasks=2
                Launched reduce tasks=1
                Total megabyte-seconds taken by all map tasks=8678400
                Total megabyte-seconds taken by all reduce tasks=3588096
                Total time spent by all map tasks (ms)=8475
                Total time spent by all maps in occupied slots (ms)=8475
                Total time spent by all reduce tasks (ms)=3504
                Total time spent by all reduces in occupied slots (ms)=3504
                Total vcore-seconds taken by all map tasks=8475
                Total vcore-seconds taken by all reduce tasks=3504
        Map-Reduce Framework
                CPU time spent (ms)=2450
                Combine input records=0
                Combine output records=0
                Failed Shuffles=0
                GC time elapsed (ms)=367
                Input split bytes=336
                Map input records=282
                Map output bytes=1737
                Map output materialized bytes=2313
                Map output records=282
                Merged Map outputs=2
                Physical memory (bytes) snapshot=651063296
                Reduce input groups=12
                Reduce input records=282
                Reduce output records=12
                Reduce shuffle bytes=2313
                Shuffled Maps =2
                Spilled Records=564
                Total committed heap usage (bytes)=679477248
                Virtual memory (bytes) snapshot=4099473408
        Shuffle Errors
                BAD_ID=0
                CONNECTION=0
                IO_ERROR=0
                WRONG_LENGTH=0
                WRONG_MAP=0
                WRONG_REDUCE=0
Streaming final output from hdfs:///user/root/tmp/mrjob/gamesbreakdown.root.20170905.135908.241472/output...
"10"    15
"11"    14
"12"    12
"13"    4
"2"     9
"3"     28
"4"     44
"5"     34
"6"     48
"7"     31
"8"     28
"9"     15
Removing HDFS temp directory hdfs:///user/root/tmp/mrjob/gamesbreakdown.root.20170905.135908.241472...
Removing temp directory /tmp/gamesbreakdown.root.20170905.135908.241472...

Now lets add another reducer step to sort the counts of the ratings.

from mrjob.job import MRJob
from mrjob.step import MRStep

class GamesBreakdownUpdate(MRJob):
    def steps(self):
        return [
            MRStep(mapper=self.mapper_get_ratings,
                   reducer=self.reducer_count_ratings),
            MRStep(reducer=self.reducer_sorted_output)
        ]

    def mapper_get_ratings(self, _, line):
        (index, name, grade, publisher, reader_rating, number_of_votes, publish_year, total_grade) = line.split('\t') 
        yield reader_rating, 1

    def reducer_count_ratings(self, key, values):
        yield str(sum(values)).zfill(2), key

    def reducer_sorted_output(self, count, ratings):
        for rating in ratings:
            yield rating, count

if __name__ == '__main__':
    GamesBreakdownUpdate.run()
[[email protected] gamedata]# python gamesbreakdownupdate.py nesgamedata.csv
No configs found; falling back on auto-configuration
Creating temp directory /tmp/gamesbreakdownupdate.root.20170905.142738.885314
Running step 1 of 2...
Running step 2 of 2...
Streaming final output from /tmp/gamesbreakdownupdate.root.20170905.142738.885314/output...
"13"    "04"
"2"     "09"
"12"    "12"
"11"    "14"
"10"    "15"
"9"     "15"
"3"     "28"
"8"     "28"
"7"     "31"
"5"     "34"
"4"     "44"
"6"     "48"
Removing temp directory /tmp/gamesbreakdownupdate.root.20170905.142738.885314...

That concludes my first experiments with Hadoop and MapReduce. Next step: using Pig or Spark to calculate similar statistics without all the overhead.