You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2013/09/23 17:56:17 UTC
svn commit: r1525632 [2/3] - in /pig/trunk: ./
src/org/apache/pig/impl/builtin/ src/org/apache/pig/impl/streaming/
src/org/apache/pig/impl/util/ src/org/apache/pig/scripting/
src/org/apache/pig/scripting/streaming/
src/org/apache/pig/scripting/streamin...
Added: pig/trunk/src/python/streaming/controller.py
URL: http://svn.apache.org/viewvc/pig/trunk/src/python/streaming/controller.py?rev=1525632&view=auto
==============================================================================
--- pig/trunk/src/python/streaming/controller.py (added)
+++ pig/trunk/src/python/streaming/controller.py Mon Sep 23 15:56:16 2013
@@ -0,0 +1,338 @@
+#! /usr/bin/env python
+import sys
+import os
+import logging
+
+from datetime import datetime
+try:
+ from dateutil import parser
+ USE_DATEUTIL = True
+except ImportError:
+ USE_DATEUTIL = False
+
+from pig_util import write_user_exception, udf_logging
+
+FIELD_DELIMITER = ','
+TUPLE_START = '('
+TUPLE_END = ')'
+BAG_START = '{'
+BAG_END = '}'
+MAP_START = '['
+MAP_END = ']'
+MAP_KEY = '#'
+PARAMETER_DELIMITER = '\t'
+PRE_WRAP_DELIM = '|'
+POST_WRAP_DELIM = '_'
+NULL_BYTE = "-"
+END_RECORD_DELIM = '|_\n'
+END_RECORD_DELIM_LENGTH = len(END_RECORD_DELIM)
+
+WRAPPED_FIELD_DELIMITER = PRE_WRAP_DELIM + FIELD_DELIMITER + POST_WRAP_DELIM
+WRAPPED_TUPLE_START = PRE_WRAP_DELIM + TUPLE_START + POST_WRAP_DELIM
+WRAPPED_TUPLE_END = PRE_WRAP_DELIM + TUPLE_END + POST_WRAP_DELIM
+WRAPPED_BAG_START = PRE_WRAP_DELIM + BAG_START + POST_WRAP_DELIM
+WRAPPED_BAG_END = PRE_WRAP_DELIM + BAG_END + POST_WRAP_DELIM
+WRAPPED_MAP_START = PRE_WRAP_DELIM + MAP_START + POST_WRAP_DELIM
+WRAPPED_MAP_END = PRE_WRAP_DELIM + MAP_END + POST_WRAP_DELIM
+WRAPPED_PARAMETER_DELIMITER = PRE_WRAP_DELIM + PARAMETER_DELIMITER + POST_WRAP_DELIM
+WRAPPED_NULL_BYTE = PRE_WRAP_DELIM + NULL_BYTE + POST_WRAP_DELIM
+
+TYPE_TUPLE = TUPLE_START
+TYPE_BAG = BAG_START
+TYPE_MAP = MAP_START
+
+TYPE_BOOLEAN = "B"
+TYPE_INTEGER = "I"
+TYPE_LONG = "L"
+TYPE_FLOAT = "F"
+TYPE_DOUBLE = "D"
+TYPE_BYTEARRAY = "A"
+TYPE_CHARARRAY = "C"
+TYPE_DATETIME = "T"
+TYPE_BIGINTEGER = "N"
+TYPE_BIGDECIMAL = "E"
+
+END_OF_STREAM = TYPE_CHARARRAY + "\x04" + END_RECORD_DELIM
+TURN_ON_OUTPUT_CAPTURING = TYPE_CHARARRAY + "TURN_ON_OUTPUT_CAPTURING" + END_RECORD_DELIM
+NUM_LINES_OFFSET_TRACE = int(os.environ.get('PYTHON_TRACE_OFFSET', 0))
+
+class PythonStreamingController:
+ def __init__(self, profiling_mode=False):
+ self.profiling_mode = profiling_mode
+
+ self.input_count = 0
+ self.next_input_count_to_log = 1
+
+ def main(self,
+ module_name, file_path, func_name, cache_path,
+ output_stream_path, error_stream_path, log_file_name, is_illustrate_str):
+ sys.stdin = os.fdopen(sys.stdin.fileno(), 'rb', 0)
+
+ #Need to ensure that user functions can't write to the streams we use to
+ #communicate with pig.
+ self.stream_output = os.fdopen(sys.stdout.fileno(), 'wb', 0)
+ self.stream_error = os.fdopen(sys.stderr.fileno(), 'wb', 0)
+
+ self.input_stream = sys.stdin
+ self.output_stream = open(output_stream_path, 'a')
+ sys.stderr = open(error_stream_path, 'w')
+ is_illustrate = is_illustrate_str == "true"
+
+ sys.path.append(file_path)
+ sys.path.append(cache_path)
+ sys.path.append('.')
+
+ logging.basicConfig(filename=log_file_name, format="%(asctime)s %(levelname)s %(message)s", level=udf_logging.udf_log_level)
+ logging.info("To reduce the amount of information being logged only a small subset of rows are logged at the INFO level. Call udf_logging.set_log_level_debug in pig_util to see all rows being processed.")
+
+ input_str = self.get_next_input()
+
+ try:
+ func = __import__(module_name, globals(), locals(), [func_name], -1).__dict__[func_name]
+ except:
+ #These errors should always be caused by user code.
+ write_user_exception(module_name, self.stream_error, NUM_LINES_OFFSET_TRACE)
+ self.close_controller(-1)
+
+ if is_illustrate or udf_logging.udf_log_level != logging.DEBUG:
+ #Only log output for illustrate after we get the flag to capture output.
+ sys.stdout = open("/dev/null", 'w')
+ else:
+ sys.stdout = self.output_stream
+
+ while input_str != END_OF_STREAM:
+ should_log = False
+ if self.input_count == self.next_input_count_to_log:
+ should_log = True
+ log_message = logging.info
+ self.update_next_input_count_to_log()
+ elif udf_logging.udf_log_level == logging.DEBUG:
+ should_log = True
+ log_message = logging.debug
+
+ try:
+ try:
+ if should_log:
+ log_message("Row %s: Serialized Input: %s" % (self.input_count, input_str))
+ inputs = deserialize_input(input_str)
+ if should_log:
+ log_message("Row %s: Deserialized Input: %s" % (self.input_count, unicode(inputs)))
+ except:
+ #Capture errors where the user passes in bad data.
+ write_user_exception(module_name, self.stream_error, NUM_LINES_OFFSET_TRACE)
+ self.close_controller(-3)
+
+ try:
+ func_output = func(*inputs)
+ if should_log:
+ log_message("Row %s: UDF Output: %s" % (self.input_count, unicode(func_output)))
+ except:
+ #These errors should always be caused by user code.
+ write_user_exception(module_name, self.stream_error, NUM_LINES_OFFSET_TRACE)
+ self.close_controller(-2)
+
+ output = serialize_output(func_output)
+ if should_log:
+ log_message("Row %s: Serialized Output: %s" % (self.input_count, output))
+
+ self.stream_output.write( "%s%s" % (output, END_RECORD_DELIM) )
+ except Exception as e:
+ #This should only catch internal exceptions with the controller
+ #and pig- not with user code.
+ import traceback
+ traceback.print_exc(file=self.stream_error)
+ sys.exit(-3)
+
+ sys.stdout.flush()
+ sys.stderr.flush()
+ self.stream_output.flush()
+ self.stream_error.flush()
+
+ input_str = self.get_next_input()
+
+ def get_next_input(self):
+ input_stream = self.input_stream
+ output_stream = self.output_stream
+
+ input_str = input_stream.readline()
+
+ while input_str.endswith(END_RECORD_DELIM) == False:
+ input_str += input_stream.readline()
+
+ if input_str == TURN_ON_OUTPUT_CAPTURING:
+ logging.debug("Turned on Output Capturing")
+ sys.stdout = output_stream
+ return self.get_next_input()
+
+ if input_str == END_OF_STREAM:
+ return input_str
+
+ self.input_count += 1
+
+ return input_str[:-END_RECORD_DELIM_LENGTH]
+
+ def update_next_input_count_to_log(self):
+ """
+ Want to log enough rows that you can see progress being made and see timings without wasting time logging thousands of rows.
+ Show first 10 rows, and then the first 5 rows of every order of magnitude (10-15, 100-105, 1000-1005, ...)
+ """
+ if self.next_input_count_to_log < 10:
+ self.next_input_count_to_log = self.next_input_count_to_log + 1
+ elif self.next_input_count_to_log % 10 == 5:
+ self.next_input_count_to_log = (self.next_input_count_to_log - 5) * 10
+ else:
+ self.next_input_count_to_log = self.next_input_count_to_log + 1
+
+ def close_controller(self, exit_code):
+ sys.stderr.close()
+ self.stream_error.write("\n")
+ self.stream_error.close()
+ sys.stdout.close()
+ self.stream_output.write("\n")
+ self.stream_output.close()
+ sys.exit(exit_code)
+
+def deserialize_input(input_str):
+ if len(input_str) == 0:
+ return []
+
+ return [_deserialize_input(param, 0, len(param)-1) for param in input_str.split(WRAPPED_PARAMETER_DELIMITER)]
+
+def _deserialize_input(input_str, si, ei):
+ if ei - si < 1:
+ #Handle all of the cases where you can have valid empty input.
+ if ei == si:
+ if input_str[si] == TYPE_CHARARRAY:
+ return u""
+ elif input_str[si] == TYPE_BYTEARRAY:
+ return bytearray("")
+ else:
+ raise Exception("Got input type flag %s, but no data to go with it.\nInput string: %s\nSlice: %s" % (input_str[si], input_str, input_str[si:ei+1]))
+ else:
+ raise Exception("Start index %d greater than end index %d.\nInput string: %s\n, Slice: %s" % (si, ei, input_str[si:ei+1]))
+
+ first = input_str[si]
+ schema = input_str[si+1] if first == PRE_WRAP_DELIM else first
+
+ if schema == NULL_BYTE:
+ return None
+ elif schema == TYPE_TUPLE or schema == TYPE_MAP or schema == TYPE_BAG:
+ return _deserialize_collection(input_str, schema, si+3, ei-3)
+ elif schema == TYPE_CHARARRAY:
+ return unicode(input_str[si+1:ei+1], 'utf-8')
+ elif schema == TYPE_BYTEARRAY:
+ return bytearray(input_str[si+1:ei+1])
+ elif schema == TYPE_INTEGER:
+ return int(input_str[si+1:ei+1])
+ elif schema == TYPE_LONG or schema == TYPE_BIGINTEGER:
+ return long(input_str[si+1:ei+1])
+ elif schema == TYPE_FLOAT or schema == TYPE_DOUBLE or schema == TYPE_BIGDECIMAL:
+ return float(input_str[si+1:ei+1])
+ elif schema == TYPE_BOOLEAN:
+ return input_str[si+1:ei+1] == "true"
+ elif schema == TYPE_DATETIME:
+ #Format is "yyyy-MM-ddTHH:mm:ss.SSS+00:00" or "2013-08-23T18:14:03.123+ZZ"
+ if USE_DATEUTIL:
+ return parser.parse(input_str[si+1:ei+1])
+ else:
+ #Try to use datetime even though it doesn't handle time zones properly,
+ #We only use the first 3 microsecond digits and drop time zone (first 23 characters)
+ return datetime.strptime(input_str[si+1:si+24], "%Y-%m-%dT%H:%M:%S.%f")
+ else:
+ raise Exception("Can't determine type of input: %s" % input_str[si:ei+1])
+
+def _deserialize_collection(input_str, return_type, si, ei):
+ list_result = []
+ append_to_list_result = list_result.append
+ dict_result = {}
+
+ index = si
+ field_start = si
+ depth = 0
+
+ key = None
+
+ # recurse to deserialize elements if the collection is not empty
+ if ei-si+1 > 0:
+ while True:
+ if index >= ei - 2:
+ if return_type == TYPE_MAP:
+ dict_result[key] = _deserialize_input(input_str, value_start, ei)
+ else:
+ append_to_list_result(_deserialize_input(input_str, field_start, ei))
+ break
+
+ if return_type == TYPE_MAP and not key:
+ key_index = input_str.find(MAP_KEY, index)
+ key = unicode(input_str[index+1:key_index], 'utf-8')
+ index = key_index + 1
+ value_start = key_index + 1
+ continue
+
+ if not (input_str[index] == PRE_WRAP_DELIM and input_str[index+2] == POST_WRAP_DELIM):
+ prewrap_index = input_str.find(PRE_WRAP_DELIM, index+1)
+ index = (prewrap_index if prewrap_index != -1 else end_index)
+ continue
+
+ mid = input_str[index+1]
+
+ if mid == BAG_START or mid == TUPLE_START or mid == MAP_START:
+ depth += 1
+ elif mid == BAG_END or mid == TUPLE_END or mid == MAP_END:
+ depth -= 1
+ elif depth == 0 and mid == FIELD_DELIMITER:
+ if return_type == TYPE_MAP:
+ dict_result[key] = _deserialize_input(input_str, value_start, index - 1)
+ key = None
+ else:
+ append_to_list_result(_deserialize_input(input_str, field_start, index - 1))
+ field_start = index + 3
+
+ index += 3
+
+ if return_type == TYPE_MAP:
+ return dict_result
+ elif return_type == TYPE_TUPLE:
+ return tuple(list_result)
+ else:
+ return list_result
+
+def serialize_output(output, utfEncodeAllFields=False):
+ """
+ @param utfEncodeStrings - Generally we want to utf encode only strings. But for
+ Maps we utf encode everything because on the Java side we don't know the schema
+ for maps so we wouldn't be able to tell which fields were encoded or not.
+ """
+
+ output_type = type(output)
+
+ if output is None:
+ return WRAPPED_NULL_BYTE
+ elif output_type == tuple:
+ return (WRAPPED_TUPLE_START +
+ WRAPPED_FIELD_DELIMITER.join([serialize_output(o, utfEncodeAllFields) for o in output]) +
+ WRAPPED_TUPLE_END)
+ elif output_type == list:
+ return (WRAPPED_BAG_START +
+ WRAPPED_FIELD_DELIMITER.join([serialize_output(o, utfEncodeAllFields) for o in output]) +
+ WRAPPED_BAG_END)
+ elif output_type == dict:
+ return (WRAPPED_MAP_START +
+ WRAPPED_FIELD_DELIMITER.join(['%s%s%s' % (k.encode('utf-8'), MAP_KEY, serialize_output(v, True)) for k, v in output.iteritems()]) +
+ WRAPPED_MAP_END)
+ elif output_type == bool:
+ return ("true" if output else "false")
+ elif output_type == bytearray:
+ return str(output)
+ elif output_type == datetime:
+ return output.isoformat()
+ elif utfEncodeAllFields or output_type == str or output_type == unicode:
+ #unicode is necessary in cases where we're encoding non-strings.
+ return unicode(output).encode('utf-8')
+ else:
+ return str(output)
+
+if __name__ == '__main__':
+ controller = PythonStreamingController()
+ controller.main(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4],
+ sys.argv[5], sys.argv[6], sys.argv[7], sys.argv[8])
Added: pig/trunk/src/python/streaming/pig_util.py
URL: http://svn.apache.org/viewvc/pig/trunk/src/python/streaming/pig_util.py?rev=1525632&view=auto
==============================================================================
--- pig/trunk/src/python/streaming/pig_util.py (added)
+++ pig/trunk/src/python/streaming/pig_util.py Mon Sep 23 15:56:16 2013
@@ -0,0 +1,85 @@
+import logging
+
+class udf_logging(object):
+ udf_log_level = logging.INFO
+
+ @classmethod
+ def set_log_level_error(cls):
+ cls.udf_log_level = logging.ERROR
+
+ @classmethod
+ def set_log_level_warn(cls):
+ cls.udf_log_level = logging.WARN
+
+ @classmethod
+ def set_log_level_info(cls):
+ cls.udf_log_level = logging.INFO
+
+ @classmethod
+ def set_log_level_debug(cls):
+ cls.udf_log_level = logging.DEBUG
+
+def outputSchema(schema_str):
+ def wrap(f):
+ def wrapped_f(*args):
+ return f(*args)
+ return wrapped_f
+ return wrap
+
+def write_user_exception(filename, stream_err_output, num_lines_offset_trace=0):
+ import sys
+ import traceback
+ import inspect
+ (t, v, tb) = sys.exc_info()
+ name = t.__name__
+ record_error = False
+
+ if name in ['SyntaxError', 'IndentationError']:
+ syntax_error_values = v.args
+ user_line_number = syntax_error_values[1][1] - num_lines_offset_trace
+ error_message = "%s: %s\n\tFile: %s, line %s column %s\n\t%s" % \
+ (name,
+ syntax_error_values[0],
+ syntax_error_values[1][0],
+ user_line_number,
+ syntax_error_values[1][2],
+ syntax_error_values[1][3])
+ else:
+ error_message = "%s: %s\n" % (name, v)
+ user_line_number = None
+ while 1:
+ e_file_name = tb.tb_frame.f_code.co_filename
+ if e_file_name.find(filename) > 0:
+ record_error = True
+ if not record_error:
+ if not tb.tb_next:
+ break
+ tb = tb.tb_next
+ continue
+
+ line_number = tb.tb_lineno
+ mod = inspect.getmodule(tb)
+ if mod:
+ lines, offset = inspect.getsourcelines(mod)
+ line = lines[line_number - offset - 1]
+ else:
+ #Useful to catch exceptions with an invalid module (like syntax
+ #errors)
+ lines, offset = inspect.getsourcelines(tb.tb_frame)
+ if (line_number - 1) >= len(lines):
+ line = "Unknown Line"
+ else:
+ line = lines[line_number - 1]
+
+ user_line_number = line_number - num_lines_offset_trace
+ func_name = tb.tb_frame.f_code.co_name
+ error_message += 'File %s, line %s, in %s\n\t%s\n' % \
+ (e_file_name, user_line_number, func_name, line)
+ if not tb.tb_next:
+ break
+ tb = tb.tb_next
+ if name in ['UnicodeEncodeError']:
+ error_message += "\nTo print a unicode string in your udf use encode('utf-8'). Example: \n\tprint 'Example'.encode('utf-8')"
+ if user_line_number:
+ stream_err_output.write("%s\n" % user_line_number)
+ stream_err_output.write("%s\n" % error_message)
Modified: pig/trunk/test/e2e/pig/build.xml
URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/build.xml?rev=1525632&r1=1525631&r2=1525632&view=diff
==============================================================================
--- pig/trunk/test/e2e/pig/build.xml (original)
+++ pig/trunk/test/e2e/pig/build.xml Mon Sep 23 15:56:16 2013
@@ -34,6 +34,7 @@
<property name="udf.jar" value="${udf.java.dir}/testudf.jar"/>
<property name="python.udf.dir" value="${udf.dir}/python"/>
<property name="ruby.udf.dir" value="${udf.dir}/ruby" />
+ <property name="cpython.udf.dir" value="${udf.dir}/cpython" />
<property name="params.dir" value="${basedir}/paramfiles"/>
<property name="e2e.lib.dir" value="${basedir}/lib"/>
<property name="streaming.dir" value="${basedir}/streaming"/>
@@ -142,6 +143,7 @@
<mkdir dir="${tar.dir}/libexec/PigTest/generate"/>
<mkdir dir="${tar.dir}/libexec/python"/>
<mkdir dir="${tar.dir}/libexec/ruby"/>
+ <mkdir dir="${tar.dir}/libexec/cpython"/>
<mkdir dir="${tar.dir}/lib"/>
<mkdir dir="${tar.dir}/lib/java"/>
<mkdir dir="${tar.dir}/paramfiles"/>
@@ -193,6 +195,10 @@
<fileset dir="${ruby.udf.dir}"/>
</copy>
+ <copy todir="${tar.dir}/libexec/cpython">
+ <fileset dir="${cpython.udf.dir}"/>
+ </copy>
+
<copy todir="${tar.dir}/paramfiles">
<fileset file="${params.dir}/params_3"/>
</copy>
Modified: pig/trunk/test/e2e/pig/tests/nightly.conf
URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/tests/nightly.conf?rev=1525632&r1=1525631&r2=1525632&view=diff
==============================================================================
--- pig/trunk/test/e2e/pig/tests/nightly.conf (original)
+++ pig/trunk/test/e2e/pig/tests/nightly.conf Mon Sep 23 15:56:16 2013
@@ -3681,6 +3681,189 @@ store b into ':OUTPATH:';\,
]
},
{
+ 'name' => 'StreamingPythonUDFs',
+ 'tests' => [
+ {
+ # test integer square
+ 'num' => 1,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478",
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate myfuncs.square(age);
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate age * age;
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test string concat and referencing function without a namespace
+ 'num' => 2,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python;
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name:chararray, age, gpa);
+b = foreach a generate concat(name) as name;
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name:chararray, age:int, gpa:double);
+b = foreach a generate CONCAT(name, name);
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test long and float square, plus two references to the same UDF with different schemas
+ 'num' => 3,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'floatpostprocess' => 1,
+ 'delimiter' => ' ',
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:long, gpa:double);
+b = foreach a generate myfuncs.square(age) as age:long, myfuncs.squareDouble(gpa) as gpa:double;
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate age * age, ((double) ROUND((gpa * gpa)*10000)) / 10000.0;
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test bytearray
+ 'num' => 4,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name:chararray, age, gpa);
+b = foreach a generate myfuncs.byteconcat(name);
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age, gpa);
+b = foreach a generate CONCAT(name, name);
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test complex types
+ 'num' => 5,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+b = foreach a generate flatten(myfuncs.complexTypes(m, t, b)) as (mm, mt, mb);
+c = foreach b generate mm#'name', mt.$0, mb.$0;
+store c into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+b = foreach a generate SIZE(m#'name'), t.$2, b.$2;
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test null input and output
+ 'num' => 6,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studentnulltab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate myfuncs.square(age);
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studentnulltab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate age * age;
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test functions that call other functions and include other files
+ 'num' => 7,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate myfuncs.redirect(age);
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate age * age;
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test that functions with same names resolve correctly across name spaces
+ 'num' => 8,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+register ':SCRIPTHOMEPATH:/cpython/morepythonudfs.py' using streaming_python as morefuncs;
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate myfuncs.square(age), morefuncs.square(age);
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = foreach a generate age * age, age * age * age;
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test that functions with same names resolve correctly across name spaces
+ 'num' => 9,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = group a by name;
+c = foreach b generate group, myfuncs.count(a);
+store c into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name, age:int, gpa:double);
+b = group a by name;
+c = foreach b generate group, COUNT(a);
+store c into ':OUTPATH:';\,
+ },
+ {
+ # test that functions with same names resolve correctly across name spaces
+ 'num' => 10,
+ 'pig' => q\
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/allscalar10k' as (name:chararray, age:int, gpa:double, instate:boolean);
+b = foreach a generate name, myfuncs.adjustgpa(gpa, instate);
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/allscalar10k' using PigStorage() as (name:chararray, age:int, gpa:double, instate:chararray);
+b = foreach a generate name, ((double)ROUND((instate=='true'?gpa:gpa+1)*10000)) / 10000.0;
+store b into ':OUTPATH:';\,
+ },
+ {
+ # test that functions with same names resolve correctly across name spaces
+ 'num' => 11,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studenttab10k' as (name:chararray, age:int, gpa:double);
+b = foreach a generate name, myfuncs.isretired(age);
+store b into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name:chararray, age:int, gpa:double);
+b = foreach a generate name, (age>=60?'true':'false');
+store b into ':OUTPATH:';\,
+ },
+ {
+ # udf which returns an array
+ 'num' => 12,
+ 'ignore23' => 'Streaming UDF does not work for Hadoop 2. See PIG-3478"
+ 'pig' => q\
+register ':SCRIPTHOMEPATH:/cpython/scriptingudf.py' using streaming_python as myfuncs;
+a = load ':INPATH:/singlefile/studenttab10k' as (name:chararray, age:chararray, gpa:chararray);
+b = foreach a generate CONCAT(CONCAT(age, ' '), gpa) as sentence;
+c = foreach b generate flatten(myfuncs.tokenize(sentence));
+store c into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+a = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name:chararray, age:chararray, gpa:chararray);
+b = load ':INPATH:/singlefile/studenttab10k' using PigStorage() as (name:chararray, age:chararray, gpa:chararray);
+c = foreach a generate age;
+d = foreach b generate gpa;
+e = union c, d;
+store e into ':OUTPATH:';\,
+ }
+ ]
+ },
+ {
'name' => 'Native',
'tests' => [
{
Added: pig/trunk/test/e2e/pig/udfs/cpython/morepythonudfs.py
URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/udfs/cpython/morepythonudfs.py?rev=1525632&view=auto
==============================================================================
--- pig/trunk/test/e2e/pig/udfs/cpython/morepythonudfs.py (added)
+++ pig/trunk/test/e2e/pig/udfs/cpython/morepythonudfs.py Mon Sep 23 15:56:16 2013
@@ -0,0 +1,22 @@
+############################################################################
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from pig_util import outputSchema
+
+@outputSchema("notsquare:int")
+def square(num):
+ return num * num * num
+
Added: pig/trunk/test/e2e/pig/udfs/cpython/scriptingudf.py
URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/udfs/cpython/scriptingudf.py?rev=1525632&view=auto
==============================================================================
--- pig/trunk/test/e2e/pig/udfs/cpython/scriptingudf.py (added)
+++ pig/trunk/test/e2e/pig/udfs/cpython/scriptingudf.py Mon Sep 23 15:56:16 2013
@@ -0,0 +1,94 @@
+############################################################################
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from pig_util import outputSchema
+
+@outputSchema("as:int")
+def square(num):
+ if num == None:
+ return None
+ return ((num) * (num))
+
+@outputSchema("as:double")
+def squareDouble(num):
+ if num == None:
+ return None
+ return ((num) * (num))
+
+@outputSchema("as:int")
+def redirect(num):
+ return square(num)
+
+@outputSchema("word:chararray")
+def concat(word):
+ return word + word
+
+@outputSchema("word:bytearray")
+def byteconcat(word):
+ return word + word
+
+@outputSchema("t:tuple(outm:map[], outt:tuple(gpa:double, age:int, name:chararray), outb:{t:(gpa:double, age:int, name:chararray)})")
+def complexTypes(m, t, b):
+ outm = {}
+ if m == None:
+ outm = None
+ else:
+ for k, v in m.iteritems():
+ outm[k] = len(v)
+
+ outb = []
+ if b == None:
+ outb = None
+ else:
+ for r in b:
+ tup = (r[2], r[1], r[0])
+ outb.append(tup)
+
+ if t == None:
+ outt = None
+ else:
+ outt = (t[2], t[1], t[0])
+
+ return (outm, outt, outb)
+
+@outputSchema("cnt:long")
+def count(bag):
+ cnt = 0
+ for r in bag:
+ cnt += 1
+ return cnt
+
+@outputSchema("gpa:double")
+def adjustgpa(gpa, instate):
+ if instate == None:
+ return None
+ elif instate:
+ return gpa
+ else:
+ return gpa+1
+
+@outputSchema("retired:boolean")
+def isretired(age):
+ if age == None:
+ return None
+ elif age>=60:
+ return True
+ else:
+ return False
+
+@outputSchema("words:{(word:chararray)}")
+def tokenize(sentence):
+ return [ (w,) for w in sentence.split(" ") ]
Modified: pig/trunk/test/excluded-tests-23
URL: http://svn.apache.org/viewvc/pig/trunk/test/excluded-tests-23?rev=1525632&r1=1525631&r2=1525632&view=diff
==============================================================================
--- pig/trunk/test/excluded-tests-23 (original)
+++ pig/trunk/test/excluded-tests-23 Mon Sep 23 15:56:16 2013
@@ -1 +1,2 @@
**/TestHBaseStorage.java
+**/TestStreamingUDF.java
Added: pig/trunk/test/org/apache/pig/impl/builtin/TestStreamingUDF.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/impl/builtin/TestStreamingUDF.java?rev=1525632&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/impl/builtin/TestStreamingUDF.java (added)
+++ pig/trunk/test/org/apache/pig/impl/builtin/TestStreamingUDF.java Mon Sep 23 15:56:16 2013
@@ -0,0 +1,303 @@
+package org.apache.pig.impl.builtin;
+
+import static org.apache.pig.builtin.mock.Storage.resetData;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.test.MiniCluster;
+import org.apache.pig.test.Util;
+import org.joda.time.DateTime;
+import org.junit.Test;
+
+public class TestStreamingUDF {
+ private static PigServer pigServerLocal = null;
+ private static PigServer pigServerMapReduce = null;
+
+ private TupleFactory tf = TupleFactory.getInstance();
+ private static MiniCluster cluster = MiniCluster.buildCluster();
+
+ @Test
+ public void testPythonUDF_onCluster() throws Exception {
+ pigServerMapReduce = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+
+ String[] pythonScript = {
+ "from pig_util import outputSchema",
+ "@outputSchema(\'c:chararray\')",
+ "def py_func(one,two):",
+ " return one + two"
+ };
+
+ Util.createLocalInputFile("pyfile_mr.py", pythonScript);
+
+ String[] input = {
+ "field10\tfield11",
+ "field20\tfield21"
+ };
+ Util.createLocalInputFile("input_mr", input);
+ Util.copyFromLocalToCluster(cluster, "input_mr", "input_mr");
+
+ pigServerMapReduce.registerQuery("REGISTER 'pyfile_mr.py' USING streaming_python AS pf;");
+ pigServerMapReduce.registerQuery("A = LOAD 'input_mr' as (c1:chararray, c2:chararray);");
+ pigServerMapReduce.registerQuery("B = FOREACH A generate pf.py_func(c1, c2);");
+
+ Iterator<Tuple> iter = pigServerMapReduce.openIterator("B");
+ assertTrue(iter.hasNext());
+ Tuple actual0 = iter.next();
+ assertTrue(iter.hasNext());
+ Tuple actual1 = iter.next();
+
+ Tuple expected0 = tf.newTuple("field10field11");
+ Tuple expected1 = tf.newTuple("field20field21");
+
+ assertEquals(expected0, actual0);
+ assertEquals(expected1, actual1);
+ }
+
+ @Test
+ public void testPythonUDF() throws Exception {
+ pigServerLocal = new PigServer(ExecType.LOCAL);
+
+ String[] pythonScript = {
+ "from pig_util import outputSchema",
+ "@outputSchema(\'c:chararray\')",
+ "def py_func(one,two):",
+ " return one + two"
+ };
+ Util.createLocalInputFile( "pyfile.py", pythonScript);
+
+
+ Data data = resetData(pigServerLocal);
+ Tuple t0 = tf.newTuple(2);
+ t0.set(0, "field10");
+ t0.set(1, "field11");
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "field20");
+ t1.set(1, "field21");
+ data.set("testTuples", "c1:chararray,c2:chararray", t0, t1);
+
+ pigServerLocal.registerQuery("REGISTER 'pyfile.py' USING streaming_python AS pf;");
+ pigServerLocal.registerQuery("A = LOAD 'testTuples' USING mock.Storage();");
+ pigServerLocal.registerQuery("B = FOREACH A generate pf.py_func(c1, c2);");
+ pigServerLocal.registerQuery("STORE B INTO 'out' USING mock.Storage();");
+
+ Tuple expected0 = tf.newTuple("field10field11");
+ Tuple expected1 = tf.newTuple("field20field21");
+
+ List<Tuple> out = data.get("out");
+ assertEquals(expected0, out.get(0));
+ assertEquals(expected1, out.get(1));
+ }
+
+ @Test
+ public void testPythonUDF_withNewline() throws Exception {
+ pigServerLocal = new PigServer(ExecType.LOCAL);
+
+ String[] pythonScript = {
+ "from pig_util import outputSchema",
+ "@outputSchema(\'c:chararray\')",
+ "def py_func(one,two):",
+ " return '%s\\n%s' % (one,two)"
+ };
+ Util.createLocalInputFile( "pyfileNL.py", pythonScript);
+
+
+ Data data = resetData(pigServerLocal);
+ Tuple t0 = tf.newTuple(2);
+ t0.set(0, "field10");
+ t0.set(1, "field11");
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "field20");
+ t1.set(1, "field21");
+ data.set("testTuplesNL", "c1:chararray,c2:chararray", t0, t1);
+
+ pigServerLocal.registerQuery("REGISTER 'pyfileNL.py' USING streaming_python AS pf;");
+ pigServerLocal.registerQuery("A = LOAD 'testTuplesNL' USING mock.Storage();");
+ pigServerLocal.registerQuery("B = FOREACH A generate pf.py_func(c1, c2);");
+ pigServerLocal.registerQuery("STORE B INTO 'outNL' USING mock.Storage();");
+
+ Tuple expected0 = tf.newTuple("field10\nfield11");
+ Tuple expected1 = tf.newTuple("field20\nfield21");
+
+ List<Tuple> out = data.get("outNL");
+ assertEquals(expected0, out.get(0));
+ assertEquals(expected1, out.get(1));
+ }
+
+ @Test
+ public void testPythonUDF__withBigInteger() throws Exception {
+ pigServerLocal = new PigServer(ExecType.LOCAL);
+
+ String[] pythonScript = {
+ "from pig_util import outputSchema",
+ "@outputSchema(\'biout:biginteger\')",
+ "def py_func(bi):",
+ " return bi"
+ };
+ Util.createLocalInputFile( "pyfile_bi.py", pythonScript);
+
+
+ Data data = resetData(pigServerLocal);
+ Tuple t0 = tf.newTuple(new BigInteger("123456789012345678901234567890"));
+ Tuple t1 = tf.newTuple(new BigInteger("9123456789012345678901234567890"));
+ data.set("testBiTuples", "bi:biginteger", t0, t1);
+
+ pigServerLocal.registerQuery("REGISTER 'pyfile_bi.py' USING streaming_python AS pf;");
+ pigServerLocal.registerQuery("A = LOAD 'testBiTuples' USING mock.Storage();");
+ pigServerLocal.registerQuery("B = FOREACH A generate pf.py_func(bi);");
+ pigServerLocal.registerQuery("STORE B INTO 'bi_out' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bi_out");
+ assertEquals(t0, out.get(0));
+ assertEquals(t1, out.get(1));
+ }
+
+ @Test
+ public void testPythonUDF__withBigDecimal() throws Exception {
+ pigServerLocal = new PigServer(ExecType.LOCAL);
+
+ String[] pythonScript = {
+ "from pig_util import outputSchema",
+ "@outputSchema(\'bdout:bigdecimal\')",
+ "def py_func(bd):",
+ " return bd"
+ };
+ Util.createLocalInputFile( "pyfile_bd.py", pythonScript);
+
+ Data data = resetData(pigServerLocal);
+ Tuple t0 = tf.newTuple(new BigDecimal("123456789012345678901234567890.12345"));
+ Tuple t1 = tf.newTuple(new BigDecimal("9123456789012345678901234567890.12345"));
+ data.set("testBdTuples", "bd:bigdecimal", t0, t1);
+
+ pigServerLocal.registerQuery("REGISTER 'pyfile_bd.py' USING streaming_python AS pf;");
+ pigServerLocal.registerQuery("A = LOAD 'testBdTuples' USING mock.Storage();");
+ pigServerLocal.registerQuery("B = FOREACH A generate pf.py_func(bd);");
+ pigServerLocal.registerQuery("STORE B INTO 'bd_out' USING mock.Storage();");
+
+ //We lose precision when we go to python.
+ List<Tuple> out = data.get("bd_out");
+ Float e0 = ((BigDecimal)t0.get(0)).floatValue();
+ Float e1 = ((BigDecimal)t1.get(0)).floatValue();
+ assertEquals(e0, ((BigDecimal) out.get(0).get(0)).floatValue(), 0.1);
+ assertEquals(e1, ((BigDecimal) out.get(1).get(0)).floatValue(), 0.1);
+ }
+
+ @Test
+ public void testPythonUDF__withDateTime() throws Exception {
+ pigServerLocal = new PigServer(ExecType.LOCAL);
+
+ String[] pythonScript = {
+ "from pig_util import outputSchema",
+ "@outputSchema(\'d:datetime\')",
+ "def py_func(dt):",
+ " return dt"
+ };
+ Util.createLocalInputFile( "pyfile_dt.py", pythonScript);
+
+
+ Data data = resetData(pigServerLocal);
+ Tuple t0 = tf.newTuple(new DateTime());
+ Tuple t1 = tf.newTuple(new DateTime());
+ data.set("testDateTuples", "d:datetime", t0, t1);
+
+ pigServerLocal.registerQuery("REGISTER 'pyfile_dt.py' USING streaming_python AS pf;");
+ pigServerLocal.registerQuery("A = LOAD 'testDateTuples' USING mock.Storage();");
+ pigServerLocal.registerQuery("B = FOREACH A generate pf.py_func(d);");
+ pigServerLocal.registerQuery("STORE B INTO 'date_out' USING mock.Storage();");
+
+ List<Tuple> out = data.get("date_out");
+ assertEquals(t0, out.get(0));
+ assertEquals(t1, out.get(1));
+ }
+
+ @Test
+ public void testPythonUDF__allTypes() throws Exception {
+ pigServerLocal = new PigServer(ExecType.LOCAL);
+
+ String[] pythonScript = {
+ "# -*- coding: utf-8 -*-",
+ "from pig_util import outputSchema",
+ "import sys",
+ "",
+ "@outputSchema('tuple_output:tuple(nully:chararray, inty:int, longy:long, floaty:float, doubly:double, chararrayy:chararray, utf_chararray_basic_string:chararray, utf_chararray_unicode:chararray, bytearrayy:bytearray)')",
+ "def get_tuple_output():",
+ " result = (None, 32, 1000000099990000L, 32.0, 3200.1234678509, 'Some String', 'Hello\\u2026Hello', u'Hello\\u2026Hello', b'Some Byte Array')",
+ " return result",
+ "",
+ "@outputSchema('tuple_output:tuple(nully:chararray, inty:int, longy:long, floaty:float, doubly:double, chararrayy:chararray, utf_chararray_basic_string:chararray, utf_chararray_unicode:chararray, bytearrayy:bytearray)')",
+ "def crazy_tuple_identity(ct):",
+ " print ct",
+ " return ct",
+ "",
+ "@outputSchema('mappy:map[]')",
+ "def add_map():",
+ " return {u'Weird\\u2026Name' : u'Weird \\u2026 Value',",
+ " 'SomeNum' : 32,",
+ " 'Simple Name' : 'Simple Value' }",
+ "",
+ "",
+ "@outputSchema('silly:chararray')",
+ "def silly(silly_word):",
+ " print silly_word.encode('utf-8')",
+ " return silly_word",
+ "",
+ "",
+ "@outputSchema('final_output:bag{t:tuple(user_id:chararray,age:int,tuple_output:tuple(nully:chararray,inty:int,longy:long,floaty:float,doubly:double,chararrayy:chararray,utf_chararray_basic_string:chararray,utf_chararray_unicode:chararray,bytearrayy:bytearray),mappy:map[],silly:chararray)}')",
+ "def bag_identity(bag):",
+ " return bag"
+ };
+ Util.createLocalInputFile("allfile.py", pythonScript);
+
+
+ Data data = resetData(pigServerLocal);
+ Tuple t0 = tf.newTuple(2);
+ t0.set(0, "user1");
+ t0.set(1, 10);
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "user2");
+ t1.set(1, 11);
+ data.set("userTuples", "user_id:chararray,age:int", t0, t1);
+
+
+ pigServerLocal.registerQuery("REGISTER 'allfile.py' USING streaming_python AS pf;");
+ pigServerLocal.registerQuery("users = LOAD 'userTuples' USING mock.Storage();");
+ pigServerLocal.registerQuery("crazy_tuple = FOREACH users GENERATE user_id, age, pf.get_tuple_output();");
+ pigServerLocal.registerQuery("crazy_tuple_fun = FOREACH crazy_tuple GENERATE user_id, age, pf.crazy_tuple_identity(tuple_output);");
+ pigServerLocal.registerQuery("crazy_tuple_with_map = FOREACH crazy_tuple_fun GENERATE user_id, age, tuple_output, pf.add_map(), pf.silly('\u2026');");
+ pigServerLocal.registerQuery("crazy_group = GROUP crazy_tuple_with_map BY (age);");
+ pigServerLocal.registerQuery("out = FOREACH crazy_group GENERATE group, pf.bag_identity(crazy_tuple_with_map);");
+
+ pigServerLocal.registerQuery("STORE out INTO 'all_out' USING mock.Storage();");
+
+ List<Tuple> out = data.get("all_out");
+
+ /*
+ * Expected output for first tuple.
+ * (10,
+ * {(user1,10,(,32,1000000099990000,32.0,3200.12346785,Some String,Hello\u2026Hello,HelloHello,Some Byte Array),
+ * [Simple Name#Simple Value,SomeNum#32,WeirdName#Weird Value],)
+ * })
+ */
+
+ //Get Bag
+ DataBag bag = (DataBag) out.get(0).get(1);
+ assertEquals(1, bag.size());
+
+ //Get First Bag Tuple
+ Tuple innerTuple = bag.iterator().next();
+ assertEquals(5, innerTuple.size());
+
+ //Check one field in innermost tuple
+ assertEquals("Hello\\u2026Hello", ((Tuple) innerTuple.get(2)).get(6));
+ }
+}
Added: pig/trunk/test/org/apache/pig/impl/streaming/TestPigStreamingUDF.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/impl/streaming/TestPigStreamingUDF.java?rev=1525632&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/impl/streaming/TestPigStreamingUDF.java (added)
+++ pig/trunk/test/org/apache/pig/impl/streaming/TestPigStreamingUDF.java Mon Sep 23 15:56:16 2013
@@ -0,0 +1,397 @@
+package org.apache.pig.impl.streaming;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.pig.builtin.ToDate;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.DefaultBagFactory;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.data.WritableByteArray;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+import org.joda.time.DateTime;
+import org.joda.time.format.ISODateTimeFormat;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class TestPigStreamingUDF {
+ PigStreamingUDF ps = new PigStreamingUDF();
+ TupleFactory tf = TupleFactory.getInstance();
+
+ @Test
+ public void testSerialize__nullTuple() throws IOException {
+ byte[] expectedOutput = "|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(null)));
+ }
+
+ @Test
+ public void testSerialize__emptyTuple() throws IOException {
+ Tuple t = tf.newTuple(0);
+ byte[] expectedOutput = "|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__record() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, "12\n34\n");
+ byte[] expectedOutput = "C12\n34\n|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__string() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, "1234");
+ byte[] expectedOutput = "C1234|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__emptyString() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, "");
+ byte[] expectedOutput = "C|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__nullString() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, null);
+ byte[] expectedOutput = "|-_|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__boolean() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, false);
+ byte[] expectedOutput = "Bfalse|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__datetime() throws IOException {
+ DateTime dt = new DateTime();
+ Tuple t = tf.newTuple(1);
+ t.set(0, dt);
+ byte[] expectedOutput = ("T" + ISODateTimeFormat.dateTime().print(dt) +"|_\n").getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__multifieldTuple() throws IOException {
+ Tuple t = tf.newTuple(2);
+ t.set(0, 1234);
+ t.set(1, "sam");
+ byte[] expectedOutput = "I1234|\t_Csam|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__multifieldTupleWithNull() throws IOException {
+ Tuple t = tf.newTuple(3);
+ t.set(0, 1234);
+ t.set(1, "sam");
+ t.set(2, null);
+ byte[] expectedOutput = "I1234|\t_Csam|\t_|-_|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__nestedTuple() throws IOException {
+ Tuple t = tf.newTuple(2);
+ Tuple nestedTuple = tf.newTuple(2);
+ nestedTuple.set(0, "sam");
+ nestedTuple.set(1, "oes");
+ t.set(0, nestedTuple);
+ t.set(1, "basil");
+ byte[] expectedOutput = "|(_Csam|,_Coes|)_|\t_Cbasil|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__bag() throws IOException {
+ Tuple t = tf.newTuple(1);
+ Tuple t1 = tf.newTuple(2);
+ Tuple t2 = tf.newTuple(2);
+ List<Tuple> bagTuples = new ArrayList<Tuple>();
+ bagTuples.add(t1);
+ bagTuples.add(t2);
+ t1.set(0, "A");
+ t1.set(1, "B");
+ t2.set(0, 1);
+ t2.set(1, 2);
+ DataBag b = DefaultBagFactory.getInstance().newDefaultBag(bagTuples);
+ t.set(0,b);
+ byte[] expectedOutput = "|{_|(_CA|,_CB|)_|,_|(_I1|,_I2|)_|}_|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__map() throws IOException {
+ Tuple t =tf.newTuple(1);
+ Map<String, String> m = new HashMap<String, String>();
+ m.put("A", "B");
+ m.put("C", "D");
+ t.set(0,m);
+ byte[] expectedOutput = "|[_CA#CB|,_CC#CD|]_|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__complex_map() throws IOException {
+ Tuple t = tf.newTuple(1);
+ Map<String, Object> inner_map = new HashMap<String, Object>();
+ inner_map.put("A", 1);
+ inner_map.put("B", "E");
+
+ Map<String, Object> outer_map = new HashMap<String, Object>();
+ outer_map.put("C", "F");
+ outer_map.put("D", inner_map);
+
+ t.set(0,outer_map);
+ byte[] expectedOutput = "|[_CD#|[_CA#I1|,_CB#CE|]_|,_CC#CF|]_|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+ @Test
+ public void testSerialize__allDataTypes() throws IOException {
+ Tuple t = tf.newTuple(8);
+ t.set(0, null);
+ t.set(1, true);
+ t.set(2, 2);
+ t.set(3, 3L);
+ t.set(4, 4.0f);
+ t.set(5, 5.0d);
+ t.set(6, new DataByteArray("six"));
+ t.set(7, "seven");
+ byte[] expectedOutput = "|-_|\t_Btrue|\t_I2|\t_L3|\t_F4.0|\t_D5.0|\t_Asix|\t_Cseven|_\n".getBytes();
+ Assert.assertTrue(assertEquals(expectedOutput, ps.serializeToBytes(t)));
+ }
+
+
+ @Test
+ public void testDeserialize__newline() throws IOException {
+ byte[] input = "12\n34\n|_".getBytes();
+ FieldSchema schema = new FieldSchema("", DataType.CHARARRAY);
+ PigStreamingUDF sp = new PigStreamingUDF(schema);
+
+ Tuple out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple("12\n34\n"), out);
+ }
+
+ @Test
+ public void testDeserialize__string() throws IOException {
+ byte[] input = "1234|_".getBytes();
+ FieldSchema schema = new FieldSchema("", DataType.CHARARRAY);
+ PigStreamingUDF sp = new PigStreamingUDF(schema);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple("1234"), out);
+ }
+
+ @Test
+ public void testDeserialize__emptyString() throws IOException {
+ byte[] input = "|_".getBytes();
+ FieldSchema schema = new FieldSchema("", DataType.CHARARRAY);
+ PigStreamingUDF sp = new PigStreamingUDF(schema);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(""), out);
+ }
+
+ @Test
+ public void testDeserialize__nullString() throws IOException {
+ byte[] input = "|-_|_".getBytes();
+ FieldSchema schema = new FieldSchema("", DataType.CHARARRAY);
+ PigStreamingUDF sp = new PigStreamingUDF(schema);
+
+ Tuple expectedOutput = tf.newTuple(1);
+ expectedOutput.set(0, null);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(expectedOutput, out);
+ }
+
+ @Test
+ public void testDeserialize__boolean() throws IOException {
+ byte[] input = "true|_".getBytes();
+ FieldSchema schema = new FieldSchema("", DataType.BOOLEAN);
+ PigStreamingUDF sp = new PigStreamingUDF(schema);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(Boolean.TRUE), out);
+ }
+
+ @Test
+ public void testDeserialize__datetime() throws IOException {
+ String dateString = "2013-08-22T09:22:08.784004";
+ byte[] input = (dateString + "|_").getBytes();
+ FieldSchema schema = new FieldSchema("", DataType.DATETIME);
+ PigStreamingUDF sp = new PigStreamingUDF(schema);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(ToDate.extractDateTime(dateString)), out);
+ }
+
+ @Test
+ public void testDeserialize__multifieldTuple() throws IOException {
+ byte[] input = "|(_1234|,_sam|)_|_".getBytes();
+ FieldSchema f1 = new FieldSchema("", DataType.INTEGER);
+ FieldSchema f2 = new FieldSchema("", DataType.CHARARRAY);
+ List<FieldSchema> fsl = new ArrayList<FieldSchema>();
+ fsl.add(f1);
+ fsl.add(f2);
+ Schema schema = new Schema(fsl);
+
+ FieldSchema fs = new FieldSchema("", schema, DataType.TUPLE);
+ Tuple expectedOutput = tf.newTuple(2);
+ expectedOutput.set(0, 1234);
+ expectedOutput.set(1, "sam");
+ PigStreamingUDF sp = new PigStreamingUDF(fs);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(expectedOutput), out);
+ }
+
+ @Test
+ public void testDeserialize__nestedTuple() throws IOException {
+ byte[] input = "|(_|(_sammy|,_oes|)_|,_1234|,_sam|)_|_".getBytes();
+ FieldSchema f1Inner = new FieldSchema("", DataType.CHARARRAY);
+ FieldSchema f2Inner = new FieldSchema("", DataType.CHARARRAY);
+ List<FieldSchema> fslInner = new ArrayList<FieldSchema>();
+ fslInner.add(f1Inner);
+ fslInner.add(f2Inner);
+ Schema schemaInner = new Schema(fslInner);
+
+ FieldSchema f1 = new FieldSchema("", schemaInner, DataType.TUPLE);
+ FieldSchema f2 = new FieldSchema("", DataType.INTEGER);
+ FieldSchema f3 = new FieldSchema("", DataType.CHARARRAY);
+ List<FieldSchema> fsl = new ArrayList<FieldSchema>();
+ fsl.add(f1);
+ fsl.add(f2);
+ fsl.add(f3);
+ Schema schema = new Schema(fsl);
+ FieldSchema fs = new FieldSchema("", schema, DataType.TUPLE);
+ PigStreamingUDF sp = new PigStreamingUDF(fs);
+
+ Tuple expectedOutputInner = tf.newTuple(2);
+ expectedOutputInner.set(0, "sammy");
+ expectedOutputInner.set(1, "oes");
+
+ Tuple expectedOutput = tf.newTuple(3);
+ expectedOutput.set(0, expectedOutputInner);
+ expectedOutput.set(1, 1234);
+ expectedOutput.set(2, "sam");
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(expectedOutput), out);
+ }
+
+ @Test
+ public void testDeserialize__bag() throws IOException {
+ byte[] input = "|{_|(_A|,_1|)_|,_|(_B|,_2|)_|}_|_".getBytes();
+ FieldSchema f1Inner = new FieldSchema("", DataType.CHARARRAY);
+ FieldSchema f2Inner = new FieldSchema("", DataType.INTEGER);
+ List<FieldSchema> fslInner = new ArrayList<FieldSchema>();
+ fslInner.add(f1Inner);
+ fslInner.add(f2Inner);
+ Schema schemaInner = new Schema(fslInner);
+ FieldSchema fsInner = new FieldSchema("", schemaInner, DataType.TUPLE);
+
+ List<FieldSchema> fsl = new ArrayList<FieldSchema>();
+ fsl.add(fsInner);
+ Schema schema = new Schema(fsl);
+
+ FieldSchema fs = new FieldSchema("", schema, DataType.BAG);
+ PigStreamingUDF sp = new PigStreamingUDF(fs);
+
+
+ Tuple expectedOutputInner1 = tf.newTuple(2);
+ expectedOutputInner1.set(0, "A");
+ expectedOutputInner1.set(1, 1);
+
+ Tuple expectedOutputInner2 = tf.newTuple(2);
+ expectedOutputInner2.set(0, "B");
+ expectedOutputInner2.set(1, 2);
+
+ List<Tuple> tuples = new ArrayList<Tuple>();
+ tuples.add(expectedOutputInner1);
+ tuples.add(expectedOutputInner2);
+ DataBag expectedOutput = DefaultBagFactory.getInstance().newDefaultBag(tuples);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(expectedOutput), out);
+ }
+
+ @Test
+ public void testDeserialize__map() throws IOException {
+ byte[] input = "|[_A#B|,_C#D|]_|_".getBytes();
+ FieldSchema fs = new FieldSchema("", DataType.MAP);
+ PigStreamingUDF sp = new PigStreamingUDF(fs);
+
+
+ Map<String, String> expectedOutput = new HashMap<String, String>();
+ expectedOutput.put("A", "B");
+ expectedOutput.put("C", "D");
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(expectedOutput), out);
+ }
+
+ @Test
+ public void testDeserialize__bug() throws Exception {
+ byte[] input = "|(_|-_|,_32|,_987654321098765432|,_987654321098765432|)_|_".getBytes();
+
+ FieldSchema f1 = new FieldSchema("", DataType.CHARARRAY);
+ FieldSchema f2 = new FieldSchema("", DataType.INTEGER);
+ FieldSchema f3 = new FieldSchema("", DataType.LONG);
+ FieldSchema f4 = new FieldSchema("", DataType.LONG);
+
+ List<FieldSchema> fsl = new ArrayList<FieldSchema>();
+ fsl.add(f1);
+ fsl.add(f2);
+ fsl.add(f3);
+ fsl.add(f4);
+ Schema schema = new Schema(fsl);
+ FieldSchema fs = new FieldSchema("", schema, DataType.TUPLE);
+ PigStreamingUDF sp = new PigStreamingUDF(fs);
+
+
+ Tuple expectedOutput1 = tf.newTuple(4);
+ expectedOutput1.set(0, null);
+ expectedOutput1.set(1, 32);
+ expectedOutput1.set(2, 987654321098765432L);
+ expectedOutput1.set(3, 987654321098765432L);
+
+ Object out = sp.deserialize(input, 0, input.length);
+ Assert.assertEquals(tf.newTuple(expectedOutput1), out);
+ }
+
+ private boolean assertEquals(byte[] expected, WritableByteArray wba) {
+ byte[] data = wba.getData();
+ if (expected.length != wba.getLength()) {
+ return false;
+ }
+
+ for (int i = 0; i < expected.length; i++) {
+ if (expected[i] != data[i]) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+}
Added: pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUDFOutputHandler.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUDFOutputHandler.java?rev=1525632&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUDFOutputHandler.java (added)
+++ pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUDFOutputHandler.java Mon Sep 23 15:56:16 2013
@@ -0,0 +1,56 @@
+package org.apache.pig.impl.streaming;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.io.UnsupportedEncodingException;
+
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.io.BufferedPositionedInputStream;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+
+
+@RunWith(JUnit4.class)
+public class TestStreamingUDFOutputHandler {
+ TupleFactory tf = TupleFactory.getInstance();
+
+ @Test
+ public void testGetValue__noNewLine() throws Exception{
+ FieldSchema fs = new FieldSchema("", DataType.CHARARRAY);
+ String data = "a|_\n";
+
+ PigStreamingUDF deserializer = new PigStreamingUDF(fs);
+ OutputHandler outty = new StreamingUDFOutputHandler(deserializer);
+ outty.bindTo(null, getIn(data), 0, 0);
+
+ Tuple t = outty.getNext();
+
+ Assert.assertEquals(tf.newTuple("a"), t);
+ }
+
+ @Test
+ public void testGetValue__embeddedNewLine() throws Exception{
+ FieldSchema fs = new FieldSchema("", DataType.CHARARRAY);
+ String data = "abc\ndef\nghi\njkl|_\n";
+
+ PigStreamingUDF deserializer = new PigStreamingUDF(fs);
+ OutputHandler outty = new StreamingUDFOutputHandler(deserializer);
+ outty.bindTo(null, getIn(data), 0, 0);
+
+ Tuple t = outty.getNext();
+
+ Assert.assertEquals(tf.newTuple("abc\ndef\nghi\njkl"), t);
+ }
+
+ private BufferedPositionedInputStream getIn(String input) throws UnsupportedEncodingException {
+ InputStream stream = new ByteArrayInputStream(input.getBytes("UTF-8"));
+ BufferedPositionedInputStream result = new BufferedPositionedInputStream(stream);
+ return result;
+ }
+}
Added: pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUtil.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUtil.java?rev=1525632&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUtil.java (added)
+++ pig/trunk/test/org/apache/pig/impl/streaming/TestStreamingUtil.java Mon Sep 23 15:56:16 2013
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.impl.streaming;
+
+import static org.apache.pig.PigConfiguration.PIG_STREAMING_ENVIRONMENT;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+import org.apache.commons.lang.RandomStringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+import com.google.common.collect.Maps;
+
+public class TestStreamingUtil {
+ private static final Random r = new Random(100L);
+
+ @Test
+ public void testAddJobConfToEnv() {
+ StringBuilder streamingEnv = null;
+ Configuration conf = new Configuration();
+ Map<String, String> all = Maps.newHashMap();
+ for (int i = 0; i < 10000; i++) {
+ String key = RandomStringUtils.randomAlphanumeric(10);
+ String value = RandomStringUtils.randomAlphanumeric(10);
+ all.put(key, value);
+ }
+ Map<String, String> toInclude = Maps.newHashMap();
+ for (Map.Entry<String, String> entry : all.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ conf.set(key, value);
+ if (r.nextDouble() < .5) {
+ toInclude.put(key, value);
+ if (streamingEnv == null) {
+ streamingEnv = new StringBuilder();
+ } else {
+ streamingEnv.append(",");
+ }
+ streamingEnv.append(key);
+ }
+ }
+ conf.set(PIG_STREAMING_ENVIRONMENT, streamingEnv.toString());
+ Map<String, String> env = new HashMap<String, String>();
+ StreamingUtil.addJobConfToEnvironment(conf, env);
+
+ for (Map.Entry<String, String> entry : env.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ String value2 = toInclude.remove(key);
+ assertEquals("Key ["+key+"] should be set in environment", value, value2);
+ }
+ assertTrue("There should be no remaining pairs in the included map",toInclude.isEmpty());
+ }
+}
Added: pig/trunk/test/org/apache/pig/test/TestPigStreaming.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestPigStreaming.java?rev=1525632&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestPigStreaming.java (added)
+++ pig/trunk/test/org/apache/pig/test/TestPigStreaming.java Mon Sep 23 15:56:16 2013
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.pig.builtin.PigStreaming;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.DefaultBagFactory;
+import org.apache.pig.data.DefaultTupleFactory;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+
+@RunWith(JUnit4.class)
+public class TestPigStreaming extends TestCase {
+ PigStreaming ps = new PigStreaming();
+ TupleFactory tf = TupleFactory.getInstance();
+
+ @Test
+ public void testSerialize__emptyTuple() throws IOException {
+ Tuple t = tf.newTuple(0);
+ byte[] expectedOutput = "".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__string() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, "1234");
+ byte[] expectedOutput = "1234\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__emptyString() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, "");
+ byte[] expectedOutput = "\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__nullString() throws IOException {
+ Tuple t = tf.newTuple(1);
+ t.set(0, null);
+ byte[] expectedOutput = "\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__multifieldTuple() throws IOException {
+ Tuple t = tf.newTuple(2);
+ t.set(0, 1234);
+ t.set(1, "sam");
+ byte[] expectedOutput = "1234\tsam\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__multifieldTupleWithNull() throws IOException {
+ Tuple t = tf.newTuple(3);
+ t.set(0, 1234);
+ t.set(1, "sam");
+ t.set(2, null);
+ byte[] expectedOutput = "1234\tsam\t\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__nestedTuple() throws IOException {
+ Tuple t = tf.newTuple(2);
+ Tuple nestedTuple = tf.newTuple(2);
+ nestedTuple.set(0, "sam");
+ nestedTuple.set(1, "oes");
+ t.set(0, nestedTuple);
+ t.set(1, "basil");
+ byte[] expectedOutput = "(sam,oes)\tbasil\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__bag() throws IOException {
+ Tuple t = tf.newTuple(1);
+ Tuple t1 = tf.newTuple(2);
+ Tuple t2 = tf.newTuple(2);
+ List<Tuple> bagTuples = new ArrayList<Tuple>();
+ bagTuples.add(t1);
+ bagTuples.add(t2);
+ t1.set(0, "A");
+ t1.set(1, "B");
+ t2.set(0, 1);
+ t2.set(1, 2);
+ DataBag b = DefaultBagFactory.getInstance().newDefaultBag(bagTuples);
+ t.set(0,b);
+ byte[] expectedOutput = "{(A,B),(1,2)}\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__map() throws IOException {
+ Tuple t = tf.newTuple(1);
+ Map<String, String> m = new HashMap<String, String>();
+ m.put("A", "B");
+ m.put("C", "D");
+ t.set(0,m);
+ byte[] expectedOutput = "[A#B,C#D]\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__complex_map() throws IOException {
+ Tuple t = tf.newTuple(1);
+ Map<String, Object> inner_map = new HashMap<String, Object>();
+ inner_map.put("A", 1);
+ inner_map.put("B", "E");
+
+ Map<String, Object> outer_map = new HashMap<String, Object>();
+ outer_map.put("C", "F");
+ outer_map.put("D", inner_map);
+
+ t.set(0,outer_map);
+ byte[] expectedOutput = "[D#[A#1,B#E],C#F]\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testSerialize__allDataTypes() throws IOException {
+ Tuple t = tf.newTuple(8);
+ t.set(0, null);
+ t.set(1, true);
+ t.set(2, 2);
+ t.set(3, 3L);
+ t.set(4, 4.0f);
+ t.set(5, 5.0d);
+ t.set(6, new DataByteArray("six"));
+ t.set(7, "seven");
+ byte[] expectedOutput = "\ttrue\t2\t3\t4.0\t5.0\tsix\tseven\n".getBytes();
+ byte[] output = ps.serialize(t);
+ Assert.assertArrayEquals(expectedOutput, output);
+ }
+
+ @Test
+ public void testDeserialize__string() throws IOException {
+ byte[] input = "1234".getBytes();
+ Tuple out = ps.deserialize(input);
+ Assert.assertEquals(tf.newTuple(new DataByteArray("1234")), out);
+ }
+
+ @Test
+ public void testDeserialize__emptyString() throws IOException {
+ byte[] input = "".getBytes();
+ Tuple expectedOutput = tf.newTuple(1);
+ expectedOutput.set(0, null);
+
+ Tuple out = ps.deserialize(input);
+
+ Assert.assertEquals(expectedOutput, out);
+ }
+}