Other developers tutorials

• 'General information'
• 'HIVE tutorial'
• 'R tutorial'

from pymongo import MongoClient

def mongo():

    try:
        print 'Connecting to the Mongo client'
        mongo_con = MongoClient( <host>, <port> )
        mongo_con[ <database> ].authenticate( <username>, <password> )
        print 'Succesfully connected!'

    except Exception, e:
        print 'There was some error: %s' % e

    return mongo_con

con = mongo()

HBase connection and usage

Read and write in the HBase database from Python scripts is permited thanks to the happybase library.

This is an example of connection:

import happybase

hbase = happybase.Connection(self.config['app']['hbase']['host'], self.config['app']['hbase']['port'])
hbase.open()

The generic HBase connection function is defined in modules/__init__.py. This function must be always used to define the context['clients']['hbase'] variable in the modules.

In the following list there are some typical HBase processes executed from Python. Remember that the advanced queries to HBase tables are always performed by HIVE, but simple processes like, create or delete a table, or to insert or remove some specific keys, are done using the happybase library. In this examples, consider that the context['clients']['hbase'] is the hbase variable.

Some typical HBase processes executed from Python are:

1. Table creation An HBase table 'ot000' is created. Their unique column family is 'results'.

hbase.create_table( 'ot000', {'results': dict()} )

2. Table delete An HBase table 'ot000' is deleted.

hbase.delete_table( 'ot000', disable=True )

3. List of existing tables Generate a string list with the name of all existing tables in HBase

hbase.tables()

3. Insert a row Insert in the key '25110~0000000000' of table 'ot000' some new column values in the results column family.

hbase_table = hbase.table('ot000')
hbase_table.put( '25110~0000000000', {'results:column1': '23', 'results:column2': 'postalCode'} )

4. Remove a specific key Remove the row with key '25110~0000000000' of table 'ot000'

hbase_table = hbase.table('ot000')
hbase_table.delete('25110~0000000000')

The happybase documentation for further information can be found in http://happybase.readthedocs.org/en/latest/api.html

Finally, the 'load_output_to_hbase' function, defined in /lib/utils.py calls the Python Map Reduce script which transform and load to HBase all the rows contained in the temporary results text files temporally stored in HDFS.

load_output_to_hbase(context, input, table, sep=",") - Arguments:

• context: the context dictionary of the module.
• input: HDFS text file or folder containing results to be loaded into HBase.
• table: Dictionary with this structure {'name':<>,'fields':[<>],'cf':<>}. The 'name' key of the dictionary is the HBase table where the results have to be loaded. The 'fields' key of the dictionary is the list of column names for each of the input text file columns (Remember to not consider the first column of the file, because it is the results key column). Finally, the 'cf' key of the dictionary is the name of the column family of the whole 'fields' list.
• sep: Column separator of the input text file.

Create a basic Map Reduce job

To explain the creation of a new Map Reduce job using Python, a very simple example will be considered.

Imagine we have an input text file in HDFS (In path /tmp/test_mrjob) with lines composed by the monthly consumption, the postal code and the country code for every contract of a company separated by 't'. First three lines of input file:

412.15\t25110\tES
234.19\t25006\tES
768.85\t08080\tES

The job will consist to calculate the average monthly consumption for every country code and postal code combination (<postalCode>~<countryCode>").

In order to execute this job, the steps to follow are:

1. Create a Python class defining the mapper, the reducer, and all the MRJob functions needed for the Map Reduce job. This class can be defined in an external file, for example, in a script called hadoop_job.py inside the directory of a module (e.g. /modules/ot000/). The main structure of this file will be:

# Library imports (All the libraries for every function used in the MRjob have to be imported here)
from mrjob.job import MRJob
from json import load

class example_mrjob(MRJob):
    # Mapper initialization to load global configurations for the mapper
    def mapper_init(self):

        # Load the config variables
        fn = glob.glob('*.json')
        self.config = load(open(fn[0]))

        # All the connections to clients (HBase, MongoDB, HDFS,...) and assignation of needed global variables for the mapper have to be defined here.

    # Reducer initialization to load global configurations for the mapper
    def reducer_init(self):

        # Load the config variables
        fn = glob.glob('*.json')
        self.config = load(open(fn[0]))

        # All the connections to clients (HBase, MongoDB, HDFS,...) and assignation of needed global variables for the reducer have to be defined here.

    # Mapper definition
    def mapper(self, _, line):

        # Convert the original string to a strings list separated by '\t'
        line = line.split('\t')

        # Val variable definition
        val = line[self.config['value_column']]

        # Key variable definition
        key = '%s~%s' % (line[self.config['postalCode_column']], line[self.config['countryCode_column']])

        # Emit the key - value to the reducer. All the keys with the same name will be treated by the same reducer process.
        yield key,val

    # Reducer definition
    def reducer(self,key,values):

        # Sum all the values and count the number of objects in values list
        c = 0
        s = 0.0
        for item in values:
            s += float(item)
            c += 1

        # Calculate the average monthly consumption
        avg = s/float(c)

        # Emit the key - value to the output file
        yield key, avg


# Run the MRjob when call the example_mrjob
if __name__ == '__main__':
    example_mrjob.run()

2. In the tasks.py of the module you are developing, insert the following code to execute the MRjob defined in the class called 'example_mrjob()' inside the script hadoop_job.py.

# Import the example_mrjob class
from hadoop_job import example_mrjob

# Create temporary file to upload configuration variables to the Map Reduce job
f = NamedTemporaryFile(delete=False, suffix='.json')
f.write(JSONdumps(context['config']))
f.close()

# Create the hadoop job instance adding file location to be uploaded
mr_job = example_mrjob(args=['-r', 'hadoop', '/tmp/test_mrjob', '--file', f.name, '--output-dir', '/tmp/output_test_mrjob', '--python-archive', path.dirname(lib.__file__)])
with mr_job.make_runner() as runner:
    try:
        runner.run()
    except Exception,e:
        f.unlink(f.name)
        raise Exception('Error running MRJob process using hadoop: %s' % e)
f.unlink(f.name)

3. Once executed, check the correction of the results in HDFS (/tmp/output_test_mrjob).