hive tips: convert default separator

hive's default separator is like '^A'.
to convert '^A' to ',', use below sed script.

/opt/hadoop/bin/hadoop dfs -cat /user/test/* | sed 's/[Ctrl-V][Ctrl-A]/,/g' > ${outputfile}

Realtime Big Data at Facebook with Hadoop and HBase

Tokyo HBase Meetup - Realtime Big Data at Facebook with Hadoop and HBase (ja)

View more presentations from tatsuya6502

hello, pig

This is my first time to touch pig which is one of hadoop-related project.

yaboo@maniac:~$ pig -x local

2011-06-22 23:45:32,781 [main] INFO  org.apache.pig.Main - Logging error messages to: /home/yaboo/pig_1308753932779.log
2011-06-22 23:45:32,849 [main] INFO  org.apache.pig.backend.hadoop.executionengine.HExecutionEngine - Connecting to hadoop file system at: file:///

grunt> A = load '/etc/passwd' using PigStorage(':'); # read /etc/password and sep is ':'

grunt> B = foreach A generate $0 as id; # get first column

grunt> dump B; # output result to stdout
 
2011-06-22 23:46:40,493 [main] INFO  org.apache.pig.tools.pigstats.ScriptState - Pig features used in the script: UNKNOWN
2011-06-22 23:46:40,493 [main] INFO  org.apache.pig.backend.hadoop.executionengine.HExecutionEngine - pig.usenewlogicalplan is set to true. New logical plan will be used.
...
2011-06-22 23:46:47,276 [main] INFO  org.apache.hadoop.mapreduce.lib.input.FileInputFormat - Total input paths to process : 1
2011-06-22 23:46:47,277 [main] INFO  org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil - Total input paths to process : 1
(root)
(daemon)
(bin)
(sys)
(sync)
(games)
(man)
(lp)
(mail)
(news)
(uucp)
(proxy)
(www-data)
(backup)
(list)
(irc)
(gnats)
(nobody)
(libuuid)
(syslog)
(messagebus)
(avahi-autoipd)
(avahi)
(couchdb)
(speech-dispatcher)
(usbmux)
(haldaemon)
(kernoops)
(pulse)
(rtkit)
(saned)
(hplip)
(gdm)
(yaboo)
(sshd)
(hadoop)

grunt>store B into 'id.out'; # stored B into id.out directory

how to make jar file for custom Hive UDF

I tried to make my original Hive User Defined Function (Hive UDF). But it was bit difficult for me to compile java file correctly, because I was a beginner of JAVA and did not know much about JAVA compile options. For compiling JAVA file and making JAR file for Hive UDF, I implemented a shell script which is almost totally copied from here 8^)

script for compiling JAVA file and making JAR file

This code is written by sh script. If you want to use bash version, you can get source code from here

#!/bin/sh

if [ $# -ne 1 ]; then
    echo "Usage: $0 "
    exit 1
fi

CNAME=${1%.java}
JARNAME=$CNAME.jar
JARDIR=/home/yaboo/$CNAME
CLASSPATH=$(ls $HIVE_HOME/lib/hive-serde-*.jar):$(ls $HIVE_HOME/lib/hive-exec-*.jar):$(ls $HADOOP_HOME/hadoop-core-*.jar)
echo "javac -classpath $CLASSPATH -d $JARDIR/ $1 && jar -cf $JARNAME -C $JARDIR/ . && tell $1"

tell() {
    echo
    echo "$1 successfully compiled."
    echo
}

mkdir -p $JARDIR
javac -classpath $CLASSPATH -d $JARDIR/ $1 && jar -cf $JARNAME -C $JARDIR/ . && tell $1

custom UDF

Here, I show custom UDF named Lower. This code is copied from official Hive Wiki

package com.example.hive.udf;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

public final class Lower extends UDF {
    public Text evaluate(final Text s) {
        if (s == null) {
            return null;
        }
        return new Text (s.toString().toLowerCase());
    }
}

Usage of script

> sh compile.sh Lower.java
Lower.java successfully compiled.
> ls Lower*
Lower.jar Lower.java Lower

hadoop-streaming: inner join

hadoop-streaming HOWTO
step1. implement mapper and reducer
step2. chmod a+x <mapper/reducer>
step3. execute like below:
> bin/hadoop jar contrib/streaming/hadoop-*-streaming.jar -file /home/hadoop/mapper.py -mapper /home/hadoop/mapper.py -file /home/hadoop/reducer.py -reducer /home/hadoop/reducer.py -input inputdir/* -output outputdir

mapper.py

#!/usr/bin/env python                                                          

import sys
import itertools

def read_input(file, separator=','):
    for line in file:
        yield line.strip().split(separator)

def main():
    rows = read_input(sys.stdin)

    for row in rows:
        if row[0] != None:
            mapper_output = get_convert_func(row[0])
            mapper_output(row)

def get_convert_func(fid):
    """ return function for convert each data """

    if fid == '001':
        return convert001
    elif fid == '002':
        return convert002
    else:
        return None

def convert001(row, separator='\t'):
    if len(row) != 6:
        return None

    fid = row[0]
    k1 = row[1]
    k2 = row[2]
    k3 = row[3]
    k4 = row[4]
    v1 = row[5]

    K = get_key_or_value_string(k1, k2, k3, k4)
    V = get_key_or_value_string(fid, v1)
    print "%s%s%s" % (K, separator, V)

def convert002(row, separator='\t'):
    if len(row) != 7:
        return None

    fid = row[0]
    k1 = row[1]
    k2 = row[2]
    k3 = row[3]
    k4 = row[4]
    v1 = row[5]
    v2 = row[6]

    K = get_key_or_value_string(k1, k2, k3, k4)
    V = get_key_or_value_string(fid, v1, v2)
    print "%s%s%s" % (K, separator, V)


def get_key_or_value_string(*args):
    return (',').join(args)


if __name__ == "__main__":
    main()

reducer.py

#!/usr/bin/env python

import sys
from itertools import groupby
from operator import itemgetter

def read_mapper_output(file, separator='\t'):
    for row in file:
        yield row.strip().split(separator, 1)

def main(separator='\t'):
    data = read_mapper_output(sys.stdin, separator=separator)

    for K, group in groupby(data, itemgetter(0)):
        L = ['-1', '-1', '-1', '-1', '-1', '-1', '-1']
        L[0], L[1], L[2], L[3] = get_columns_from_key(K, 4, ',')

        for K, V in group:
            columns = V.split(',')
            if columns[0] == '001':
                L[4] = columns[1] 
                L[5] = columns[2]
            elif columns[0] == '002':
                L[6] = columns[1]
            else:
                pass

        if is_healthy_output(L):
            print get_string_from_list(L)

def is_healthy_output(row):
    """ return True if output row does not have '-1' column """
    if [c for c in row if c == '-1'] == []:
        return True
    else:
        return False

def get_string_from_list(L):
    return (',').join(L)

def get_columns_from_key(key, length, separator=','):
    L = key.split(',')
    if len(L) != length:
        return None
    else:
        return L

if __name__ == "__main__":
    main()