You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2014/05/20 17:21:28 UTC
[2/4] git commit: UDF Wrappers and Tests
UDF Wrappers and Tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/e4557ea8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/e4557ea8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/e4557ea8
Branch: refs/heads/master
Commit: e4557ea8d006f281862b71fbcee86d272aae3289
Parents: a63d3e2
Author: sblackmon <sb...@w2odigital.com>
Authored: Mon May 19 19:20:39 2014 -0500
Committer: sblackmon <sb...@w2odigital.com>
Committed: Mon May 19 19:20:39 2014 -0500
----------------------------------------------------------------------
.../streams/pig/StreamsProcessDatumExec.java | 89 ++++++++++++++++----
.../streams/pig/test/CopyThriceProcessor.java | 37 ++++++++
.../streams/pig/test/DoNothingProcessor.java | 35 ++++++++
.../streams/pig/test/PigProcessDatumTest.java | 80 ++++++++++++++++++
.../pig/test/PigProcessDocumentTest.java | 71 ++++++++++++++++
.../streams/pig/test/PigSerializerTest.java | 43 ++++++++++
.../test/resources/pigprocessdatumcopytest.pig | 4 +
.../src/test/resources/pigprocessdatumtest.pig | 4 +
.../test/resources/pigprocessdocumenttest.pig | 4 +
9 files changed, 350 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/main/java/org/apache/streams/pig/StreamsProcessDatumExec.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/main/java/org/apache/streams/pig/StreamsProcessDatumExec.java b/streams-runtimes/streams-runtime-pig/src/main/java/org/apache/streams/pig/StreamsProcessDatumExec.java
index 2f7538d..28d3516 100644
--- a/streams-runtimes/streams-runtime-pig/src/main/java/org/apache/streams/pig/StreamsProcessDatumExec.java
+++ b/streams-runtimes/streams-runtime-pig/src/main/java/org/apache/streams/pig/StreamsProcessDatumExec.java
@@ -7,10 +7,7 @@ import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.pig.EvalFunc;
import org.apache.pig.builtin.MonitoredUDF;
-import org.apache.pig.data.BagFactory;
-import org.apache.pig.data.DataBag;
-import org.apache.pig.data.Tuple;
-import org.apache.pig.data.TupleFactory;
+import org.apache.pig.data.*;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.util.UDFContext;
import org.apache.streams.core.StreamsDatum;
@@ -39,14 +36,15 @@ public class StreamsProcessDatumExec extends AliasableEvalFunc<DataBag> {
Preconditions.checkArgument(execArgs.length > 0);
String classFullName = execArgs[0];
Preconditions.checkNotNull(classFullName);
- String[] constructorArgs = new String[execArgs.length-1];
+ String[] prepareArgs = new String[execArgs.length-1];
ArrayUtils.remove(execArgs, 0);
- ArrayUtils.addAll(constructorArgs, execArgs);
- if( constructorArgs.length == 0 )
- streamsProcessor = StreamsComponentFactory.getProcessorInstance(Class.forName(classFullName));
- else
- streamsProcessor = StreamsComponentFactory.getProcessorInstance(Class.forName(classFullName), constructorArgs);
- streamsProcessor.prepare(null);
+ ArrayUtils.addAll(prepareArgs, execArgs);
+ streamsProcessor = StreamsComponentFactory.getProcessorInstance(Class.forName(classFullName));
+ if( execArgs.length == 1 ) {
+ streamsProcessor.prepare(null);
+ } else if( execArgs.length > 1 ) {
+ streamsProcessor.prepare(prepareArgs);
+ }
}
@Override
@@ -59,15 +57,27 @@ public class StreamsProcessDatumExec extends AliasableEvalFunc<DataBag> {
Configuration conf = UDFContext.getUDFContext().getJobConf();
- String id = getString(input, "id");
- String source = getString(input, "source");
+// I would prefer it work this way, but at the moment it doesn't
+
+// String id = getString(input, "id");
+// String source = getString(input, "source");
+// Long timestamp;
+// try {
+// timestamp = getLong(input, "timestamp");
+// } catch( Exception e ) {
+// timestamp = RFC3339Utils.parseUTC(getString(input, "timestamp")).getMillis();
+// }
+// String object = getString(input, "object");
+
+ String id = (String) input.get(0);
+ String source = (String) input.get(1);
Long timestamp;
try {
- timestamp = getLong(input, "timestamp");
+ timestamp = (Long) input.get(2);
} catch( Exception e ) {
- timestamp = RFC3339Utils.parseUTC(getString(input, "timestamp")).getMillis();
+ timestamp = RFC3339Utils.parseUTC((String)input.get(2)).getMillis();
}
- String object = getString(input, "object");
+ String object = (String) input.get(3);
StreamsDatum entry = new StreamsDatum(object, id, new DateTime(timestamp));
@@ -95,6 +105,51 @@ public class StreamsProcessDatumExec extends AliasableEvalFunc<DataBag> {
@Override
public Schema getOutputSchema(Schema schema) {
- return null;
+ // Check that we were passed two fields
+ String error = "Expected: id\tsource\ttimestamp\tobject";
+ if (schema.size() != 4) {
+ throw new RuntimeException(error);
+ }
+
+ try {
+ // Get the types for both columns and check them. If they are
+ // wrong, figure out what types were passed and give a good error
+ // message.
+ if (schema.getField(0).type != DataType.CHARARRAY &&
+ schema.getField(0).type != DataType.LONG) {
+ error += "Problem with id: must be CHARARRAY or LONG";
+ error += "\t(";
+ error += DataType.findTypeName(schema.getField(0).type);
+ error += ")\n";
+ throw new RuntimeException(error);
+ }
+ if (schema.getField(1).type != DataType.CHARARRAY) {
+ error += "Problem with source: must be CHARARRAY";
+ error += "\t(";
+ error += DataType.findTypeName(schema.getField(1).type);
+ error += ")\n";
+ throw new RuntimeException(error);
+ }
+ if (schema.getField(2).type != DataType.CHARARRAY &&
+ schema.getField(2).type != DataType.LONG) {
+ error += "Problem with timestamp: must be CHARARRAY or LONG";
+ error += "\t(";
+ error += DataType.findTypeName(schema.getField(2).type);
+ error += ")\n";
+ throw new RuntimeException(error);
+ }
+ if (schema.getField(3).type != DataType.CHARARRAY) {
+ error += "Problem with object: must be CHARARRAY";
+ error += "\t(";
+ error += DataType.findTypeName(schema.getField(3).type);
+ error += ")\n";
+ throw new RuntimeException(error);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(error);
+ }
+
+ // Always hand back the same schema we are passed
+ return schema;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/CopyThriceProcessor.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/CopyThriceProcessor.java b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/CopyThriceProcessor.java
new file mode 100644
index 0000000..463625a
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/CopyThriceProcessor.java
@@ -0,0 +1,37 @@
+package org.apache.streams.pig.test;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Used to Test Pig processor wrapper
+ */
+public class CopyThriceProcessor implements StreamsProcessor {
+
+ List<StreamsDatum> result;
+
+ public CopyThriceProcessor() {
+ }
+
+ @Override
+ public List<StreamsDatum> process(StreamsDatum entry) {
+ this.result = new LinkedList<StreamsDatum>();
+ result.add(entry);
+ result.add(entry);
+ result.add(entry);
+ return result;
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+
+ }
+
+ @Override
+ public void cleanUp() {
+ System.out.println("Processor clean up!");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/DoNothingProcessor.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/DoNothingProcessor.java b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/DoNothingProcessor.java
new file mode 100644
index 0000000..f99ecfc
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/DoNothingProcessor.java
@@ -0,0 +1,35 @@
+package org.apache.streams.pig.test;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Used to Test Pig processor wrapper
+ */
+public class DoNothingProcessor implements StreamsProcessor {
+
+ List<StreamsDatum> result;
+
+ public DoNothingProcessor() {
+ }
+
+ @Override
+ public List<StreamsDatum> process(StreamsDatum entry) {
+ this.result = new LinkedList<StreamsDatum>();
+ result.add(entry);
+ return result;
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+
+ }
+
+ @Override
+ public void cleanUp() {
+ System.out.println("Processor clean up!");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDatumTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDatumTest.java b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDatumTest.java
new file mode 100644
index 0000000..00db946
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDatumTest.java
@@ -0,0 +1,80 @@
+package org.apache.streams.pig.test;
+
+import org.apache.pig.pigunit.PigTest;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.twitter.processor.TwitterTypeConverter;
+import org.apache.streams.twitter.serializer.TwitterJsonTweetActivitySerializer;
+import org.apache.tools.ant.util.StringUtils;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.text.ParseException;
+import java.util.List;
+
+/**
+ * Created by sblackmon on 3/30/14.
+ */
+public class PigProcessDatumTest {
+
+ @Test
+ public void testPigDoNothingSingleDatum() throws Exception {
+ String[] args = {};
+
+ String[] input = {
+ "159475541894897679\ttwitter,statuses/user_timeline\t1384499359006\t{content:\"content\",[\"a\":1,\"b\":\"c\"}",
+ };
+
+ DoNothingProcessor processor = new DoNothingProcessor();
+
+ String doc = (String) StringUtils.split(input[0], '\t').get(3);
+ StreamsDatum inputDatum = new StreamsDatum(doc);
+ inputDatum.setId((String) StringUtils.split(input[0], '\t').get(0));
+
+ processor.prepare(null);
+
+ StreamsDatum resultDatum = processor.process(inputDatum).get(0);
+ String resultDocument = (String) resultDatum.getDocument();
+
+ String[] output = new String[1];
+ output[0] = "(159475541894897679,twitter,statuses/user_timeline,1384499359006," + resultDocument + ")";
+
+ PigTest test;
+ test = new PigTest("src/test/resources/pigprocessdatumtest.pig", args);
+ test.assertOutput("in", input, "out", output);
+
+ }
+
+ @Test
+ public void testPigCopyThriceSingleDatum() throws Exception {
+ String[] args = {};
+
+ String[] input = {
+ "159475541894897679\ttwitter,statuses/user_timeline\t1384499359006\t{content:\"content\",[\"a\":1,\"b\":\"c\"}",
+ };
+
+ CopyThriceProcessor processor = new CopyThriceProcessor();
+
+ String doc = (String) StringUtils.split(input[0], '\t').get(3);
+ StreamsDatum inputDatum = new StreamsDatum(doc);
+ inputDatum.setId((String) StringUtils.split(input[0], '\t').get(0));
+
+ processor.prepare(null);
+
+ List<StreamsDatum> resultSet = processor.process(inputDatum);
+
+ String[] output = new String[resultSet.size()];
+
+ for( int i = 0; i < output.length; i++ ) {
+ output[i] = "(159475541894897679,twitter,statuses/user_timeline,1384499359006," + resultSet.get(i).getDocument() + ")";
+ }
+
+ PigTest test;
+ test = new PigTest("src/test/resources/pigprocessdatumcopytest.pig", args);
+ test.assertOutput("in", input, "out", output);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDocumentTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDocumentTest.java b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDocumentTest.java
new file mode 100644
index 0000000..d2f77f0
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigProcessDocumentTest.java
@@ -0,0 +1,71 @@
+package org.apache.streams.pig.test;
+
+import org.apache.pig.pigunit.PigTest;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.twitter.processor.TwitterTypeConverter;
+import org.apache.streams.twitter.serializer.TwitterJsonActivitySerializer;
+import org.apache.streams.twitter.serializer.TwitterJsonTweetActivitySerializer;
+import org.apache.tools.ant.util.StringUtils;
+import org.junit.Test;
+
+/**
+ * Created by sblackmon on 3/30/14.
+ */
+public class PigProcessDocumentTest {
+
+ @Test
+ public void testPigProcessEmptyDocument() throws Exception {
+
+ String[] input = {
+ "159475541894897679\ttwitter,statuses/user_timeline\t1384499359006\t{}"
+ };
+
+ DoNothingProcessor processor = new DoNothingProcessor();
+
+ String doc = (String) StringUtils.split(input[0], '\t').get(3);
+ StreamsDatum inputDatum = new StreamsDatum(doc);
+ inputDatum.setId((String) StringUtils.split(input[0], '\t').get(0));
+
+ processor.prepare(null);
+
+ StreamsDatum resultDatum = processor.process(inputDatum).get(0);
+ String resultDocument = (String) resultDatum.getDocument();
+
+ String[] output = new String[1];
+ output[0] = "(159475541894897679,twitter,statuses/user_timeline,1384499359006,"+resultDocument+")";
+
+ PigTest test;
+ test = new PigTest("src/test/resources/pigprocessdocumenttest.pig");
+ test.assertOutput("in", input, "out", output);
+
+ }
+
+ @Test
+ public void testPigProcessJsonDocument() throws Exception {
+
+ String[] input = {
+ "159475541894897679\ttwitter,statuses/user_timeline\t1384499359006\t{content:\"content\",[\"a\":1,\"b\":\"c\"}"
+ };
+
+ DoNothingProcessor processor = new DoNothingProcessor();
+
+ String doc = (String) StringUtils.split(input[0], '\t').get(3);
+ StreamsDatum inputDatum = new StreamsDatum(doc);
+ inputDatum.setId((String) StringUtils.split(input[0], '\t').get(0));
+
+ processor.prepare(null);
+
+ StreamsDatum resultDatum = processor.process(inputDatum).get(0);
+ String resultDocument = (String) resultDatum.getDocument();
+
+ String[] output = new String[1];
+ output[0] = "(159475541894897679,twitter,statuses/user_timeline,1384499359006,"+resultDocument+")";
+
+ PigTest test;
+ test = new PigTest("src/test/resources/pigprocessdocumenttest.pig");
+ test.assertOutput("in", input, "out", output);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigSerializerTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigSerializerTest.java b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigSerializerTest.java
new file mode 100644
index 0000000..9a6d557
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/java/org/apache/streams/pig/test/PigSerializerTest.java
@@ -0,0 +1,43 @@
+package org.apache.streams.pig.test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.pigunit.PigTest;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.twitter.serializer.TwitterJsonActivitySerializer;
+import org.apache.streams.twitter.serializer.TwitterJsonTweetActivitySerializer;
+import org.apache.tools.ant.util.StringUtils;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.Iterator;
+
+/**
+ * Created by sblackmon on 3/30/14.
+ */
+public class PigSerializerTest {
+
+ @Test
+ public void testPigSerializer() throws Exception {
+
+ String[] input = {
+ "159475541894897679\ttwitter,statuses/user_timeline\t1384499359006\t{\"retweeted_status\":{\"contributors\":null,\"text\":\"The Costa Concordia cruise ship accident could be a disaster for the industry | http://t.co/M9UUNvZi (via @TIMEMoneyland)\",\"geo\":null,\"retweeted\":false,\"in_reply_to_screen_name\":null,\"possibly_sensitive\":false,\"truncated\":false,\"lang\":\"en\",\"entities\":{\"symbols\":[],\"urls\":[{\"expanded_url\":\"http://ti.me/zYyEtD\",\"indices\":[80,100],\"display_url\":\"ti.me/zYyEtD\",\"url\":\"http://t.co/M9UUNvZi\"}],\"hashtags\":[],\"user_mentions\":[{\"id\":245888431,\"name\":\"TIME Moneyland\",\"indices\":[106,120],\"screen_name\":\"TIMEMoneyland\",\"id_str\":\"245888431\"}]},\"in_reply_to_status_id_str\":null,\"id\":159470076259602432,\"source\":\"<a href=\\\"http://www.hootsuite.com\\\" rel=\\\"nofollow\\\">HootSuite<\\/a>\",\"in_reply_to_user_id_str\":null,\"favorited\":false,\"in_reply_to_status_id\":null,\"retweet_count\":71,\"create
d_at\":\"Wed Jan 18 03:00:03 +0000 2012\",\"in_reply_to_user_id\":null,\"favorite_count\":14,\"id_str\":\"159470076259602432\",\"place\":null,\"user\":{\"location\":\"\",\"default_profile\":false,\"profile_background_tile\":true,\"statuses_count\":70754,\"lang\":\"en\",\"profile_link_color\":\"1B4F89\",\"profile_banner_url\":\"https://pbs.twimg.com/profile_banners/14293310/1355243462\",\"id\":14293310,\"following\":false,\"protected\":false,\"favourites_count\":59,\"profile_text_color\":\"000000\",\"description\":\"Breaking news and current events from around the globe. Hosted by TIME staff. Tweet questions to our customer service team @TIMEmag_Service.\",\"verified\":true,\"contributors_enabled\":false,\"profile_sidebar_border_color\":\"000000\",\"name\":\"TIME.com\",\"profile_background_color\":\"CC0000\",\"created_at\":\"Thu Apr 03 13:54:30 +0000 2008\",\"default_profile_image\":false,\"followers_count\":5146268,\"profile_image_url_https\":\"https://pbs.twimg.com/profile_images/1
700796190/Picture_24_normal.png\",\"geo_enabled\":false,\"profile_background_image_url\":\"http://a0.twimg.com/profile_background_images/735228291/107f1a300a90ee713937234bb3d139c0.jpeg\",\"profile_background_image_url_https\":\"https://si0.twimg.com/profile_background_images/735228291/107f1a300a90ee713937234bb3d139c0.jpeg\",\"follow_request_sent\":false,\"entities\":{\"description\":{\"urls\":[]},\"url\":{\"urls\":[{\"expanded_url\":\"http://www.time.com\",\"indices\":[0,22],\"display_url\":\"time.com\",\"url\":\"http://t.co/4aYbUuAeSh\"}]}},\"url\":\"http://t.co/4aYbUuAeSh\",\"utc_offset\":-18000,\"time_zone\":\"Eastern Time (US & Canada)\",\"notifications\":false,\"profile_use_background_image\":true,\"friends_count\":742,\"profile_sidebar_fill_color\":\"D9D9D9\",\"screen_name\":\"TIME\",\"id_str\":\"14293310\",\"profile_image_url\":\"http://pbs.twimg.com/profile_images/1700796190/Picture_24_normal.png\",\"listed_count\":76944,\"is_translator\":false},\"coordinates\":null},\"contr
ibutors\":null,\"text\":\"RT @TIME: The Costa Concordia cruise ship accident could be a disaster for the industry | http://t.co/M9UUNvZi (via @TIMEMoneyland)\",\"geo\":null,\"retweeted\":false,\"in_reply_to_screen_name\":null,\"possibly_sensitive\":false,\"truncated\":false,\"lang\":\"en\",\"entities\":{\"symbols\":[],\"urls\":[{\"expanded_url\":\"http://ti.me/zYyEtD\",\"indices\":[90,110],\"display_url\":\"ti.me/zYyEtD\",\"url\":\"http://t.co/M9UUNvZi\"}],\"hashtags\":[],\"user_mentions\":[{\"id\":14293310,\"name\":\"TIME.com\",\"indices\":[3,8],\"screen_name\":\"TIME\",\"id_str\":\"14293310\"},{\"id\":245888431,\"name\":\"TIME Moneyland\",\"indices\":[116,130],\"screen_name\":\"TIMEMoneyland\",\"id_str\":\"245888431\"}]},\"in_reply_to_status_id_str\":null,\"id\":159475541894897679,\"source\":\"<a href=\\\"http://twitter.com/download/iphone\\\" rel=\\\"nofollow\\\">Twitter for iPhone<\\/a>\",\"in_reply_to_user_id_str\":null,\"favorited\":false,\"in_reply_to_status_id\":null,\"retwe
et_count\":71,\"created_at\":\"Wed Jan 18 03:21:46 +0000 2012\",\"in_reply_to_user_id\":null,\"favorite_count\":0,\"id_str\":\"159475541894897679\",\"place\":null,\"user\":{\"location\":\"\",\"default_profile\":false,\"profile_background_tile\":true,\"statuses_count\":5053,\"lang\":\"en\",\"profile_link_color\":\"738D84\",\"id\":27552112,\"following\":false,\"protected\":false,\"favourites_count\":52,\"profile_text_color\":\"97CEC9\",\"description\":\"\",\"verified\":false,\"contributors_enabled\":false,\"profile_sidebar_border_color\":\"A9AC00\",\"name\":\"rafael medina-flores\",\"profile_background_color\":\"C5EFE3\",\"created_at\":\"Mon Mar 30 01:21:55 +0000 2009\",\"default_profile_image\":false,\"followers_count\":963,\"profile_image_url_https\":\"https://pbs.twimg.com/profile_images/2519547938/image_normal.jpg\",\"geo_enabled\":true,\"profile_background_image_url\":\"http://a0.twimg.com/profile_background_images/167479660/trireme.jpg\",\"profile_background_image_url_https\":\"
https://si0.twimg.com/profile_background_images/167479660/trireme.jpg\",\"follow_request_sent\":false,\"entities\":{\"description\":{\"urls\":[]}},\"url\":null,\"utc_offset\":-25200,\"time_zone\":\"Mountain Time (US & Canada)\",\"notifications\":false,\"profile_use_background_image\":true,\"friends_count\":1800,\"profile_sidebar_fill_color\":\"5C4F3C\",\"screen_name\":\"rmedinaflores\",\"id_str\":\"27552112\",\"profile_image_url\":\"http://pbs.twimg.com/profile_images/2519547938/image_normal.jpg\",\"listed_count\":50,\"is_translator\":false},\"coordinates\":null}"
+ };
+
+ TwitterJsonActivitySerializer serializer = new TwitterJsonActivitySerializer();
+
+ String doc = (String) StringUtils.split(input[0], '\t').get(3);
+ String outdoc = StreamsJacksonMapper.getInstance().writeValueAsString(serializer.deserialize(doc));
+
+ String[] output = new String[1];
+ output[0] = "(159475541894897679,twitter,statuses/user_timeline,1384499359006," + outdoc + ")";
+
+ PigTest test;
+ test = new PigTest("src/test/resources/pigserializertest.pig");
+ test.assertOutput("in", input, "out", output);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumcopytest.pig
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumcopytest.pig b/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumcopytest.pig
new file mode 100644
index 0000000..63e5fa8
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumcopytest.pig
@@ -0,0 +1,4 @@
+DEFINE PROCESSOR org.apache.streams.pig.StreamsProcessDatumExec('org.apache.streams.pig.test.CopyThriceProcessor');
+in = LOAD '*' USING PigStorage('\t') AS (id: chararray, source: chararray, timestamp: long, object: chararray);
+out = FOREACH in GENERATE FLATTEN(PROCESSOR(id, source, timestamp, object));
+STORE out INTO 'target' USING PigStorage('\t');
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumtest.pig
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumtest.pig b/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumtest.pig
new file mode 100644
index 0000000..7271add
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdatumtest.pig
@@ -0,0 +1,4 @@
+DEFINE PROCESSOR org.apache.streams.pig.StreamsProcessDatumExec('org.apache.streams.pig.test.DoNothingProcessor');
+in = LOAD '*' USING PigStorage('\t') AS (id: chararray, source: chararray, timestamp: long, object: chararray);
+out = FOREACH in GENERATE FLATTEN(PROCESSOR(id, source, timestamp, object));
+STORE out INTO 'target' USING PigStorage('\t');
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e4557ea8/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdocumenttest.pig
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdocumenttest.pig b/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdocumenttest.pig
new file mode 100644
index 0000000..3685e22
--- /dev/null
+++ b/streams-runtimes/streams-runtime-pig/src/test/resources/pigprocessdocumenttest.pig
@@ -0,0 +1,4 @@
+DEFINE PROCESSOR org.apache.streams.pig.StreamsProcessDocumentExec('org.apache.streams.pig.test.DoNothingProcessor');
+in = LOAD '*' USING PigStorage('\t') AS (activityid: chararray, source: chararray, timestamp: long, object: chararray);
+out = FOREACH in GENERATE activityid, source, timestamp, PROCESSOR(object);
+STORE out INTO 'target' USING PigStorage('\t');