You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ga...@apache.org on 2009/04/23 22:52:09 UTC
svn commit: r768049 - in /hadoop/pig/trunk: CHANGES.txt
src/org/apache/pig/impl/logicalLayer/schema/SchemaUtil.java
test/org/apache/pig/test/TestSchemaUtil.java
Author: gates
Date: Thu Apr 23 20:52:09 2009
New Revision: 768049
URL: http://svn.apache.org/viewvc?rev=768049&view=rev
Log:
PIG-712: Added utility functions to create schemas for tuples and bags.
Added:
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/SchemaUtil.java
hadoop/pig/trunk/test/org/apache/pig/test/TestSchemaUtil.java
Modified:
hadoop/pig/trunk/CHANGES.txt
Modified: hadoop/pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=768049&r1=768048&r2=768049&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Thu Apr 23 20:52:09 2009
@@ -28,6 +28,9 @@
PIG-700: To automate the pig patch test process (gkesavan via sms)
+PIG-712: Added utility functions to create schemas for tuples and bags (zjffdu
+via gates).
+
BUG FIXES
PIG-733: Order by sampling dumps entire sample to hdfs which causes dfs
Added: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/SchemaUtil.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/SchemaUtil.java?rev=768049&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/SchemaUtil.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/SchemaUtil.java Thu Apr 23 20:52:09 2009
@@ -0,0 +1,261 @@
+package org.apache.pig.impl.logicalLayer.schema;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+
+/**
+ *
+ * A utility class for simplify the schema creation, especially for bag and
+ * tuple schema. Currently, it only support simple schema creation, nested tuple
+ * and bag is not supported
+ *
+ */
+
+public class SchemaUtil {
+
+ private static Set<Byte> SUPPORTED_TYPE_SET;
+
+ static {
+ SUPPORTED_TYPE_SET = new HashSet<Byte>();
+
+ SUPPORTED_TYPE_SET.add(DataType.INTEGER);
+ SUPPORTED_TYPE_SET.add(DataType.LONG);
+ SUPPORTED_TYPE_SET.add(DataType.CHARARRAY);
+ SUPPORTED_TYPE_SET.add(DataType.BOOLEAN);
+ SUPPORTED_TYPE_SET.add(DataType.BYTE);
+ SUPPORTED_TYPE_SET.add(DataType.BYTEARRAY);
+ SUPPORTED_TYPE_SET.add(DataType.DOUBLE);
+ SUPPORTED_TYPE_SET.add(DataType.FLOAT);
+ SUPPORTED_TYPE_SET.add(DataType.MAP);
+ }
+
+ /**
+ * Create a new tuple schema according the tuple name and two list: names of
+ * fields, types of fields
+ *
+ * @param tupleName
+ * @param fieldNames
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newTupleSchema(String tupleName,
+ List<String> fieldNames, List<Byte> dataTypes)
+ throws FrontendException {
+ checkParameters(fieldNames, dataTypes);
+
+ List<Schema.FieldSchema> tokenSchemas = new ArrayList<Schema.FieldSchema>();
+ for (int i = 0; i < fieldNames.size(); ++i) {
+ String name = fieldNames.get(i);
+ Byte type = dataTypes.get(i);
+ tokenSchemas.add(new Schema.FieldSchema(name, type));
+ }
+
+ Schema tupleSchema = new Schema(tokenSchemas);
+ Schema.FieldSchema tupleField = new Schema.FieldSchema(tupleName,
+ tupleSchema);
+
+ return new Schema(tupleField);
+ }
+
+ /**
+ * Create a new tuple schema according the tuple name and two arrays: names
+ * of fields, types of fields
+ *
+ * @param tupleName
+ * @param fieldNames
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newTupleSchema(String tupleName, String[] fieldNames,
+ Byte[] dataTypes) throws FrontendException {
+ return newTupleSchema(tupleName, Arrays.asList(fieldNames), Arrays
+ .asList(dataTypes));
+ }
+
+ /**
+ * Create a new tuple schema according the two list: names of fields, types
+ * of fields, the default tuple name is t.
+ *
+ * @param names
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newTupleSchema(List<String> fieldNames,
+ List<Byte> dataTypes) throws FrontendException {
+ return newTupleSchema("t", fieldNames, dataTypes);
+ }
+
+ /**
+ * Create a new tuple schema according one list: types of fields, the
+ * default names of fields are f0,f1,f2..., and the tuple name is t.
+ *
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newTupleSchema(List<Byte> dataTypes)
+ throws FrontendException {
+ List<String> names = newNames(dataTypes.size());
+ return newTupleSchema("t", names, dataTypes);
+ }
+
+ /**
+ * Create a new tuple schema according the two arrays: names of fields,
+ * types of fields, the default tuple name is t.
+ *
+ * @param names
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newTupleSchema(String[] names, Byte[] dataTypes)
+ throws FrontendException {
+ return newTupleSchema("t", Arrays.asList(names), Arrays
+ .asList(dataTypes));
+ }
+
+ /**
+ * Create a new tuple schema according one array: types of fields, the
+ * default names of fields are f0,f1,f2..., and the tuple name is t.
+ *
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newTupleSchema(Byte[] dataTypes)
+ throws FrontendException {
+ return newTupleSchema(Arrays.asList(dataTypes));
+ }
+
+ private static List<String> newNames(int size) {
+ List<String> names = new ArrayList<String>();
+ for (int i = 0; i < size; ++i) {
+ names.add("f" + i);
+ }
+ return names;
+ }
+
+ /**
+ * Create a bag schema according the bag name,tuple name and two list: name
+ * of fields, type of fields
+ *
+ * @param bagName
+ * @param tupleName
+ * @param fieldNames
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newBagSchema(String bagName, String tupleName,
+ List<String> fieldNames, List<Byte> dataTypes)
+ throws FrontendException {
+ checkParameters(fieldNames, dataTypes);
+
+ Schema tupleSchema = newTupleSchema(tupleName, fieldNames, dataTypes);
+ Schema.FieldSchema bagField = new Schema.FieldSchema(bagName,
+ tupleSchema, DataType.BAG);
+
+ return new Schema(bagField);
+ }
+
+ public static Schema newBagSchema(String bagName, String tupleName,
+ String[] fieldNames, Byte[] dataTypes) throws FrontendException {
+ return newBagSchema(bagName, tupleName, Arrays.asList(fieldNames),
+ Arrays.asList(dataTypes));
+ }
+
+ /**
+ * Create a bag schema according two list: name of fields, type of fields,
+ * and the default bag name is b, the default tuple name is t.
+ *
+ * @param names
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newBagSchema(List<String> names, List<Byte> dataTypes)
+ throws FrontendException {
+ checkParameters(names, dataTypes);
+
+ Schema tupleSchema = newTupleSchema(names, dataTypes);
+ Schema.FieldSchema bagField = new Schema.FieldSchema("b", tupleSchema,
+ DataType.BAG);
+
+ return new Schema(bagField);
+ }
+
+ /**
+ * Create a new tuple schema according one list: types of fields, the
+ * default names of fields are f0,f1,f2..., and the tuple is t, the bag name
+ * is b.
+ *
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newBagSchema(List<Byte> dataTypes)
+ throws FrontendException {
+ List<String> names = newNames(dataTypes.size());
+ return newBagSchema(names, dataTypes);
+ }
+
+ /**
+ * Create a new tuple schema according two arrays: names of field,types of
+ * fields. The default tuple name is t, and the bag is b.
+ *
+ * @param names
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newBagSchema(String[] names, Byte[] dataTypes)
+ throws FrontendException {
+ return newBagSchema(Arrays.asList(names), Arrays.asList(dataTypes));
+ }
+
+ /**
+ * Create a new tuple schema according one array: the type of fields, the
+ * tuple name is t, and the bag name is b.
+ *
+ * @param dataTypes
+ * @return
+ * @throws FrontendException
+ */
+ public static Schema newBagSchema(Byte[] dataTypes)
+ throws FrontendException {
+ return newBagSchema(Arrays.asList(dataTypes));
+ }
+
+ private static void checkDataTypes(List<Byte> dataTypes)
+ throws FrontendException {
+ for (Byte type : dataTypes) {
+ if (!SUPPORTED_TYPE_SET.contains(type)) {
+ throw new FrontendException(
+ "Currently pig do not support this kind of type using Schema:"
+ + DataType.findTypeName(type)
+ + ". You can write shema by yourself.");
+ }
+ }
+
+ }
+
+ private static void checkParameters(List<String> names, List<Byte> dataTypes)
+ throws FrontendException {
+ // TODO Auto-generated method stub
+ checkDataTypes(dataTypes);
+ if (names.size() != dataTypes.size()) {
+ throw new FrontendException(
+ "The number of names is not equal to the number of dataTypes");
+ }
+ }
+
+}
Added: hadoop/pig/trunk/test/org/apache/pig/test/TestSchemaUtil.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestSchemaUtil.java?rev=768049&view=auto
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestSchemaUtil.java (added)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestSchemaUtil.java Thu Apr 23 20:52:09 2009
@@ -0,0 +1,82 @@
+package org.apache.pig.test;
+
+import java.util.Arrays;
+
+import junit.framework.TestCase;
+
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.SchemaUtil;
+
+public class TestSchemaUtil extends TestCase {
+
+ public void testTupleSchema() {
+ try {
+ String tupleName = "mytuple";
+ String[] fieldNames = new String[] { "field_0", "field_1" };
+ Byte[] dataTypes = new Byte[] { DataType.LONG, DataType.CHARARRAY };
+
+ String expected = "{mytuple: (field_0: long,field_1: chararray)}";
+ Schema tupleSchema = SchemaUtil.newTupleSchema(tupleName,
+ fieldNames, dataTypes);
+ assertEquals(expected, tupleSchema.toString());
+
+ tupleSchema = SchemaUtil.newTupleSchema(tupleName, Arrays
+ .asList(fieldNames), Arrays.asList(dataTypes));
+ assertEquals(expected, tupleSchema.toString());
+
+ expected = "{t: (field_0: long,field_1: chararray)}";
+ tupleSchema = SchemaUtil.newTupleSchema(fieldNames, dataTypes);
+ assertEquals(expected, tupleSchema.toString());
+
+ tupleSchema = SchemaUtil.newTupleSchema(Arrays.asList(fieldNames),
+ Arrays.asList(dataTypes));
+ assertEquals(expected, tupleSchema.toString());
+
+ expected = "{t: (f0: long,f1: chararray)}";
+ tupleSchema = SchemaUtil.newTupleSchema(dataTypes);
+ assertEquals(expected, tupleSchema.toString());
+
+ tupleSchema = SchemaUtil.newTupleSchema(Arrays.asList(dataTypes));
+ assertEquals(expected, tupleSchema.toString());
+ } catch (FrontendException e) {
+ fail();
+ }
+ }
+
+ public void testBagSchema() {
+ try {
+ String bagName="mybag";
+ String tupleName = "mytuple";
+ String[] fieldNames = new String[] { "field_0", "field_1" };
+ Byte[] dataTypes = new Byte[] { DataType.LONG, DataType.CHARARRAY };
+
+ String expected = "{mybag: {mytuple: (field_0: long,field_1: chararray)}}";
+ Schema bagSchema = SchemaUtil.newBagSchema(bagName,tupleName,
+ fieldNames, dataTypes);
+ assertEquals(expected, bagSchema.toString());
+
+ bagSchema = SchemaUtil.newBagSchema(bagName,tupleName, Arrays
+ .asList(fieldNames), Arrays.asList(dataTypes));
+ assertEquals(expected, bagSchema.toString());
+
+ expected = "{b: {t: (field_0: long,field_1: chararray)}}";
+ bagSchema = SchemaUtil.newBagSchema(fieldNames, dataTypes);
+ assertEquals(expected, bagSchema.toString());
+
+ bagSchema = SchemaUtil.newBagSchema(Arrays.asList(fieldNames),
+ Arrays.asList(dataTypes));
+ assertEquals(expected, bagSchema.toString());
+
+ expected = "{b: {t: (f0: long,f1: chararray)}}";
+ bagSchema = SchemaUtil.newBagSchema(dataTypes);
+ assertEquals(expected, bagSchema.toString());
+
+ bagSchema = SchemaUtil.newBagSchema(Arrays.asList(dataTypes));
+ assertEquals(expected, bagSchema.toString());
+ } catch (FrontendException e) {
+ fail();
+ }
+ }
+}