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/12/16 21:37:15 UTC
svn commit: r891419 - in /hadoop/pig/trunk: ./
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/
contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/ ivy/
src/org/apache/pig/builtin/ src/org/apache/pig/data/ src/org/...
Author: gates
Date: Wed Dec 16 20:37:14 2009
New Revision: 891419
URL: http://svn.apache.org/viewvc?rev=891419&view=rev
Log:
PIG-760: Add a new PigStorageSchema load/store function that store schemas for text files.
Added:
hadoop/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java
hadoop/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java
hadoop/pig/trunk/src/org/apache/pig/experimental/
hadoop/pig/trunk/src/org/apache/pig/experimental/JsonMetadata.java
hadoop/pig/trunk/src/org/apache/pig/experimental/LoadMetadata.java
hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceSchema.java
hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceStatistics.java
hadoop/pig/trunk/src/org/apache/pig/experimental/StoreMetadata.java
hadoop/pig/trunk/test/org/apache/pig/test/TestResourceSchema.java
Modified:
hadoop/pig/trunk/CHANGES.txt
hadoop/pig/trunk/build.xml
hadoop/pig/trunk/ivy.xml
hadoop/pig/trunk/ivy/libraries.properties
hadoop/pig/trunk/ivy/pig.pom
hadoop/pig/trunk/src/org/apache/pig/builtin/PigStorage.java
hadoop/pig/trunk/src/org/apache/pig/data/DataType.java
hadoop/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java
Modified: hadoop/pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Wed Dec 16 20:37:14 2009
@@ -24,6 +24,9 @@
IMPROVEMENTS
+PIG-760: Add a new PigStorageSchema load/store function that
+ store schemas for text files (dvryaboy via gates)
+
PIG-1106: FR join should not spill (ankit.modi via olgan)
PIG-1147: Zebra Docs for Pig 0.6.0 (chandec via olgan)
Modified: hadoop/pig/trunk/build.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/build.xml?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/build.xml (original)
+++ hadoop/pig/trunk/build.xml Wed Dec 16 20:37:14 2009
@@ -168,7 +168,10 @@
<fileset file="${lib.dir}/${hbase.jarfile}" />
<fileset file="${lib.dir}/${hbase.test.jarfile}" />
<fileset file="${lib.dir}/${zookeeper.jarfile}"/>
- <!-- <fileset file="${lib.dir}/commons-collections-3.2.jar" /> -->
+ <fileset file="${ivy.lib.dir}/jackson-mapper-asl-${jackson.version}.jar"/>
+ <fileset file="${ivy.lib.dir}/jackson-core-asl-${jackson.version}.jar"/>
+
+ <!-- <fileset file="${lib.dir}/commons-collections-3.2.jar" /> -->
</path>
<!-- javadoc-classpath -->
@@ -422,6 +425,8 @@
<zipfileset src="${ivy.lib.dir}/junit-${junit.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jsch-${jsch.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jline-${jline.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-mapper-asl-${jackson.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-core-asl-${jackson.version}.jar" />
</jar>
<copy file="${output.jarfile}" tofile="${output.jarfile.backcompat}"/>
</target>
@@ -475,6 +480,8 @@
<zipfileset src="${ivy.lib.dir}/junit-${junit.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jsch-${jsch.version}.jar" />
<zipfileset src="${ivy.lib.dir}/jline-${jline.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-mapper-asl-${jackson.version}.jar" />
+ <zipfileset src="${ivy.lib.dir}/jackson-core-asl-${jackson.version}.jar" />
</jar>
<copy file="${output.jarfile.withouthadoop}" tofile="${output.jarfile.backcompat.withouthadoop}"/>
</target>
Added: hadoop/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java (added)
+++ hadoop/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/PigStorageSchema.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,125 @@
+/*
+ * 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.piggybank.storage;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.pig.ExecType;
+import org.apache.pig.experimental.JsonMetadata;
+import org.apache.pig.experimental.LoadMetadata;
+import org.apache.pig.experimental.StoreMetadata;
+import org.apache.pig.experimental.ResourceSchema;
+import org.apache.pig.experimental.ResourceStatistics;
+import org.apache.pig.StoreConfig;
+import org.apache.pig.backend.datastorage.DataStorage;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.backend.hadoop.datastorage.HDataStorage;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
+import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
+import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+
+/**
+ * This Load/Store Func reads/writes metafiles that allow the schema and
+ * aliases to be determined at load time, saving one from having to manually
+ * enter schemas for pig-generated datasets.
+ *
+ * It also creates a ".pig_headers" file that simply lists the delimited aliases.
+ * This is intended to make export to tools that can read files with header
+ * lines easier (just cat the header to your data).
+ *
+ * Due to StoreFunc limitations, you can only write the metafiles in MapReduce
+ * mode. You can read them in Local or MapReduce mode.
+ */
+public class PigStorageSchema extends PigStorage implements StoreMetadata {
+
+ private static final Log log = LogFactory.getLog(PigStorageSchema.class);
+
+ public PigStorageSchema() {
+ super();
+ }
+
+ public PigStorageSchema(String delim) {
+ super(delim);
+ }
+
+ @Override
+ public Schema determineSchema(String fileName, ExecType execType,
+ DataStorage storage) throws IOException {
+
+ // TODO fullPath should be retrieved ia relativeToAbsolutePath once PIG-966 is complete
+ String fullPath = FileLocalizer.fullPath(fileName, storage);
+ LoadMetadata metadataLoader = new JsonMetadata(fullPath, storage);
+ ResourceSchema resourceSchema = metadataLoader.getSchema(fullPath, null);
+ if (resourceSchema == null) {
+ return null;
+ }
+ Schema pigSchema = new Schema();
+ for (ResourceSchema.ResourceFieldSchema field : resourceSchema.getFields()) {
+ FieldSchema pigFieldSchema = DataType.determineFieldSchema(field);
+ // determineFieldSchema only sets the types. we also want the aliases.
+ // TODO this doesn't work properly for complex types
+ pigFieldSchema.alias = field.getName();
+ pigSchema.add(pigFieldSchema);
+ }
+ log.info("Loaded Schema: "+pigSchema);
+ return pigSchema;
+ }
+
+ @Override
+ public void finish() throws IOException {
+ super.finish();
+ JobConf jobConf = PigMapReduce.sJobConf;
+ if(jobConf != null){
+ StoreConfig storeConfig = MapRedUtil.getStoreConfig(jobConf);
+ DataStorage store = new HDataStorage(ConfigurationUtil.toProperties(jobConf));
+ Schema schema = storeConfig.getSchema();
+ ResourceSchema resourceSchema = new ResourceSchema(schema);
+ JsonMetadata metadataWriter = new JsonMetadata(storeConfig.getLocation(), store);
+ metadataWriter.setFieldDel(fieldDel);
+ metadataWriter.setRecordDel(recordDel);
+ metadataWriter.setSchema(resourceSchema, storeConfig.getLocation(), null);
+ }
+ }
+
+ /**
+ * @see org.apache.pig.experimental.StoreMetadata#setSchema(ResourceSchema)
+ * Does not do anything in this implementation. The finish() call writes the schema.
+ */
+ @Override
+ public void setSchema(ResourceSchema schema, String location, Configuration conf) throws IOException {
+ // n\a
+ }
+
+ /**
+ * @see org.apache.pig.experimental.StoreMetadata#setStatistics(ResourceStatistics)
+ * Does not do anything in this implementation.
+ */
+ @Override
+ public void setStatistics(ResourceStatistics stats, String location, Configuration conf) throws IOException {
+ // n\a
+ }
+}
Added: hadoop/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java (added)
+++ hadoop/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/TestPigStorageSchema.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,111 @@
+/*
+ * 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.piggybank.test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.logicalLayer.LogicalOperator;
+import org.apache.pig.impl.logicalLayer.LogicalPlan;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.test.MiniCluster;
+import org.apache.pig.test.Util;
+import org.apache.pig.test.utils.TypeCheckingTestUtil;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import junit.framework.TestCase;
+
+public class TestPigStorageSchema extends TestCase {
+
+ protected ExecType execType = ExecType.MAPREDUCE;
+
+ PigContext pigContext = new PigContext(ExecType.MAPREDUCE, new Properties());
+ Map<LogicalOperator, LogicalPlan> aliases = new HashMap<LogicalOperator, LogicalPlan>();
+ Map<OperatorKey, LogicalOperator> logicalOpTable = new HashMap<OperatorKey, LogicalOperator>();
+ Map<String, LogicalOperator> aliasOp = new HashMap<String, LogicalOperator>();
+ Map<String, String> fileNameMap = new HashMap<String, String>();
+
+ MiniCluster cluster = MiniCluster.buildCluster();
+
+ private PigServer pig;
+
+ @Before
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String origPath = FileLocalizer.fullPath("originput", pig.getPigContext());
+ if (FileLocalizer.fileExists(origPath, pig.getPigContext())) {
+ FileLocalizer.delete(origPath, pig.getPigContext());
+ }
+ Util.createInputFile(cluster, "originput",
+ new String[] {"A,1", "B,2", "C,3", "D,2",
+ "A,5", "B,5", "C,8", "A,8",
+ "D,8", "A,9"});
+
+ }
+
+ @After
+ @Override
+ protected void tearDown() throws Exception {
+ Util.deleteFile(cluster, "originput");
+ String aoutPath = FileLocalizer.fullPath("aout", pig.getPigContext());
+ if (FileLocalizer.fileExists(aoutPath, pig.getPigContext())) {
+ FileLocalizer.delete(aoutPath, pig.getPigContext());
+ }
+ }
+
+ @Test
+ public void testPigStorageSchema() throws Exception {
+ pigContext.connect();
+ String query = "a = LOAD 'originput' using org.apache.pig.piggybank.storage.PigStorageSchema() as (f1:chararray, f2:int);";
+ pig.registerQuery(query);
+ Schema origSchema = pig.dumpSchema("a");
+ pig.registerQuery("STORE a into 'aout' using org.apache.pig.piggybank.storage.PigStorageSchema();");
+
+ // aout now has a schema.
+
+ // Verify that loading a-out with no given schema produces
+ // the original schema.
+
+ pig.registerQuery("b = LOAD 'aout' using org.apache.pig.piggybank.storage.PigStorageSchema();");
+ Schema genSchema = pig.dumpSchema("b");
+ assertTrue("generated schema equals original" , Schema.equals(genSchema, origSchema, true, false));
+
+ // Verify that giving our own schema works
+ String [] aliases ={"foo", "bar"};
+ byte[] types = {DataType.INTEGER, DataType.LONG};
+ Schema newSchema = TypeCheckingTestUtil.genFlatSchema(
+ aliases,types);
+ pig.registerQuery("c = LOAD 'aout' using org.apache.pig.piggybank.storage.PigStorageSchema() as (foo:int, bar:long);");
+ Schema newGenSchema = pig.dumpSchema("c");
+ assertTrue("explicit schema overrides metadata", Schema.equals(newSchema, newGenSchema, true, false));
+
+ }
+}
Modified: hadoop/pig/trunk/ivy.xml
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/ivy.xml?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/ivy.xml (original)
+++ hadoop/pig/trunk/ivy.xml Wed Dec 16 20:37:14 2009
@@ -67,6 +67,10 @@
conf="test->default"/>
<dependency org="com.google.code.p.arat" name="rat-lib" rev="${rats-lib.version}"
conf="releaseaudit->default"/>
+ <dependency org="org.codehaus.jackson" name="jackson-mapper-asl" rev="${jackson.version}"
+ conf="compile->master"/>
+ <dependency org="org.codehaus.jackson" name="jackson-core-asl" rev="${jackson.version}"
+ conf="compile->master"/>
<!--ATM hbase, hbase-test.jar, hadoop.jar are resolved from the lib dir-->
</dependencies>
</ivy-module>
Modified: hadoop/pig/trunk/ivy/libraries.properties
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/ivy/libraries.properties?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/ivy/libraries.properties (original)
+++ hadoop/pig/trunk/ivy/libraries.properties Wed Dec 16 20:37:14 2009
@@ -33,3 +33,5 @@
rats-lib.version=0.5.1
xerces.version=1.4.4
+
+jackson.version=1.0.1
Modified: hadoop/pig/trunk/ivy/pig.pom
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/ivy/pig.pom?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/ivy/pig.pom (original)
+++ hadoop/pig/trunk/ivy/pig.pom Wed Dec 16 20:37:14 2009
@@ -83,5 +83,11 @@
<version>${junit.version}</version>
</dependency>
+ <dependency>
+ <groupId>jackson</groupId>
+ <artifactId>jackson</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+
</dependencies>
</project>
Modified: hadoop/pig/trunk/src/org/apache/pig/builtin/PigStorage.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/builtin/PigStorage.java?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/builtin/PigStorage.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/builtin/PigStorage.java Wed Dec 16 20:37:14 2009
@@ -57,15 +57,16 @@
private String signature;
long end = Long.MAX_VALUE;
- private byte recordDel = '\n';
- private byte fieldDel = '\t';
+ protected byte recordDel = '\n';
+ protected byte fieldDel = '\t';
private ArrayList<Object> mProtoTuple = null;
- private static final String UTF8 = "UTF-8";
private boolean[] mRequiredColumns = null;
private boolean mRequiredColumnsInitialized = false;
+ protected static final String UTF8 = "UTF-8";
+
public PigStorage() {}
/**
Modified: hadoop/pig/trunk/src/org/apache/pig/data/DataType.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/data/DataType.java?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/data/DataType.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/data/DataType.java Wed Dec 16 20:37:14 2009
@@ -20,13 +20,14 @@
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;
-import java.util.Set;
import org.apache.pig.PigException;
+import org.apache.pig.experimental.ResourceSchema;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
@@ -939,104 +940,132 @@
}
return true;
}
-
- /***
- * Determine the field schema of an object
- * @param o the object whose field schema is to be determined
- * @return the field schema corresponding to the object
- * @throws ExecException,FrontendException,SchemaMergeException
- */
- public static Schema.FieldSchema determineFieldSchema(Object o)
- throws ExecException, FrontendException, SchemaMergeException {
- byte dt = findType(o);
+
- switch (dt) {
+ /**
+ * Utility method that determines the schema from the passed in dataType.
+ * If the dataType is Bag or Tuple, then we need to determine the schemas inside this dataType;
+ * for this we iterate through the fields inside this field. This method works both for raw objects
+ * and ResourceSchema.ResourceFieldSchema field descriptions; the specific behavior is determined by the klass
+ * parameter.
+ * @param dataType DataType.CHARARRAY, DataType.TUPLE, and so on
+ * @param fieldIter iterator over the fields if this is a tuple or a bag
+ * @param fieldNum number of fields inside the field if a tuple
+ * @param klass should be Object or ResourceSchema.ResourceFieldSchema
+ * @return
+ * @throws ExecException
+ * @throws FrontendException
+ * @throws SchemaMergeException
+ */
+ private static Schema.FieldSchema determineFieldSchema(byte dataType, Iterator fieldIter,
+ long fieldNum, Class klass ) throws ExecException, FrontendException, SchemaMergeException {
+ switch (dataType) {
case NULL:
return new Schema.FieldSchema(null, BYTEARRAY);
case BOOLEAN:
- return new Schema.FieldSchema(null, BOOLEAN);
-
case INTEGER:
- return new Schema.FieldSchema(null, INTEGER);
-
case LONG:
- return new Schema.FieldSchema(null, LONG);
-
case FLOAT:
- return new Schema.FieldSchema(null, FLOAT);
-
case DOUBLE:
- return new Schema.FieldSchema(null, DOUBLE);
-
case BYTEARRAY:
- return new Schema.FieldSchema(null, BYTEARRAY);
-
case CHARARRAY:
- return new Schema.FieldSchema(null, CHARARRAY);
-
case MAP:
- return new Schema.FieldSchema(null, MAP);
-
+ return new Schema.FieldSchema(null, dataType);
case TUPLE: {
- Tuple t = (Tuple)o;
- long tupleSize = t.size();
- Schema schema = null;
-
- if(tupleSize != 0) {
- schema = new Schema();
- for(int i = 0; i < t.size(); ++i) {
- schema.add(determineFieldSchema(t.get(i)));
- }
+ Schema schema = null;
+ if(fieldNum != 0) {
+ schema = new Schema();
+ for(int i = 0; i < fieldNum; ++i) {
+ schema.add(determineFieldSchema(klass.cast(fieldIter.next())));
}
- return new Schema.FieldSchema(null, schema, TUPLE);
}
-
+ return new Schema.FieldSchema(null, schema, TUPLE);
+ }
+
case BAG: {
- DataBag b = (DataBag)o;
- long bagSize = b.size();
- Schema schema = null;
- Schema bagSchema = null;
-
- if(bagSize != 0) {
- Iterator<Tuple> it = b.iterator();
- ArrayList<Schema> schemas = new ArrayList<Schema>();
- while(it.hasNext()) {
- schemas.add(determineFieldSchema((Object)it.next()).schema);
- }
- schema = schemas.get(0);
- if(null == schema) {
+ Schema schema = null;
+ Schema bagSchema = null;
+
+ if(fieldNum != 0) {
+ ArrayList<Schema> schemas = new ArrayList<Schema>();
+ while (fieldIter.hasNext() ) {
+ schemas.add(determineFieldSchema(klass.cast(fieldIter.next())).schema);
+ }
+ schema = schemas.get(0);
+ if(null == schema) {
+ Schema.FieldSchema tupleFs = new Schema.FieldSchema(null, null, TUPLE);
+ bagSchema = new Schema(tupleFs);
+ bagSchema.setTwoLevelAccessRequired(true);
+ return new Schema.FieldSchema(null, bagSchema, BAG);
+ }
+ int schemaSize = schema.size();
+
+ for(int i = 1; i < schemas.size(); ++i) {
+ Schema currSchema = schemas.get(i);
+ if((null == currSchema) || (currSchema.size() != schemaSize)) {
Schema.FieldSchema tupleFs = new Schema.FieldSchema(null, null, TUPLE);
bagSchema = new Schema(tupleFs);
bagSchema.setTwoLevelAccessRequired(true);
return new Schema.FieldSchema(null, bagSchema, BAG);
}
- int schemaSize = schema.size();
-
- for(int i = 1; i < schemas.size(); ++i) {
- Schema currSchema = schemas.get(i);
- if((null == currSchema) || (currSchema.size() != schemaSize)) {
- Schema.FieldSchema tupleFs = new Schema.FieldSchema(null, null, TUPLE);
- bagSchema = new Schema(tupleFs);
- bagSchema.setTwoLevelAccessRequired(true);
- return new Schema.FieldSchema(null, bagSchema, BAG);
- }
- schema = Schema.mergeSchema(schema, currSchema, false, false, false);
- }
- Schema.FieldSchema tupleFs = new Schema.FieldSchema(null, schema, TUPLE);
- bagSchema = new Schema(tupleFs);
- // since this schema has tuple field schema which internally
- // has a list of field schemas for the actual items in the bag
- // an access to any field in the bag is a two level access
- bagSchema.setTwoLevelAccessRequired(true);
+ schema = Schema.mergeSchema(schema, currSchema, false, false, false);
}
- return new Schema.FieldSchema(null, bagSchema, BAG);
+ Schema.FieldSchema tupleFs = new Schema.FieldSchema(null, schema, TUPLE);
+ bagSchema = new Schema(tupleFs);
+ // since this schema has tuple field schema which internally
+ // has a list of field schemas for the actual items in the bag
+ // an access to any field in the bag is a two level access
+ bagSchema.setTwoLevelAccessRequired(true);
}
+ return new Schema.FieldSchema(null, bagSchema, BAG);
+ }
default: {
- int errCode = 1073;
- String msg = "Cannot determine field schema for " + o;
- throw new ExecException(msg, errCode, PigException.INPUT);
- }
+ int errCode = 1073;
+ String msg = "Cannot determine field schema";
+ throw new ExecException(msg, errCode, PigException.INPUT);
+ }
+
+ }
+ }
+
+ /***
+ * Determine the field schema of an ResourceFieldSchema
+ * @param rcFieldSchema the rcFieldSchema we want translated
+ * @return the field schema corresponding to the object
+ * @throws ExecException,FrontendException,SchemaMergeException
+ */
+ public static Schema.FieldSchema determineFieldSchema(ResourceSchema.ResourceFieldSchema rcFieldSchema)
+ throws ExecException, FrontendException, SchemaMergeException {
+ byte dt = rcFieldSchema.getType();
+ Iterator<ResourceSchema.ResourceFieldSchema> fieldIter = null;
+ long fieldNum = 0;
+ if (dt == TUPLE || dt == BAG ) {
+ fieldIter = Arrays.asList(rcFieldSchema.getSchema().getFields()).iterator();
+ fieldNum = rcFieldSchema.getSchema().getFields().length;
+ }
+ return determineFieldSchema(dt, fieldIter, fieldNum, ResourceSchema.ResourceFieldSchema.class);
+ }
+
+
+ /***
+ * Determine the field schema of an object
+ * @param o the object whose field schema is to be determined
+ * @return the field schema corresponding to the object
+ * @throws ExecException,FrontendException,SchemaMergeException
+ */
+ public static Schema.FieldSchema determineFieldSchema(Object o)
+ throws ExecException, FrontendException, SchemaMergeException {
+ byte dt = findType(o);
+ Iterator fieldIter = null;
+ long fieldNum = 0;
+ if ( dt == TUPLE ) {
+ fieldIter = ((Tuple) o).getAll().iterator();
+ fieldNum = ((Tuple) o).size();
+ } else if ( dt == BAG ) {
+ fieldNum = ((DataBag) o).size();
+ fieldIter = ((DataBag)o).iterator();
}
+ return determineFieldSchema(dt, fieldIter, fieldNum, Object.class);
}
}
Added: hadoop/pig/trunk/src/org/apache/pig/experimental/JsonMetadata.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/experimental/JsonMetadata.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/experimental/JsonMetadata.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/experimental/JsonMetadata.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,298 @@
+/*
+ * 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.experimental;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.pig.backend.datastorage.ContainerDescriptor;
+import org.apache.pig.backend.datastorage.DataStorage;
+import org.apache.pig.backend.datastorage.ElementDescriptor;
+import org.apache.pig.backend.hadoop.datastorage.HDataStorage;
+import org.apache.pig.backend.hadoop.datastorage.HDirectory;
+import org.apache.pig.backend.hadoop.datastorage.HFile;
+import org.apache.pig.backend.local.datastorage.LocalDataStorage;
+import org.apache.pig.backend.local.datastorage.LocalDir;
+import org.apache.pig.backend.local.datastorage.LocalFile;
+import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.plan.Operator;
+import org.apache.pig.impl.plan.OperatorPlan;
+import org.apache.pig.impl.plan.PlanVisitor;
+import org.codehaus.jackson.JsonGenerationException;
+import org.codehaus.jackson.JsonParseException;
+import org.codehaus.jackson.map.JsonMappingException;
+import org.codehaus.jackson.map.ObjectMapper;
+
+/**
+ * Reads and Writes metadata using JSON in metafiles next to the data.
+ *
+ *
+ */
+public class JsonMetadata implements LoadMetadata, StoreMetadata {
+
+ private static final Log log = LogFactory.getLog(JsonMetadata.class);
+
+ // These are not static+final because we may want to make these adjustable by users.
+ private String schemaFileName = ".pig_schema";
+ private String headerFileName = ".pig_header";
+ private String statFileName = ".pig_stats";
+
+ private boolean printHeaders = true;
+
+ private byte fieldDel;
+ private byte recordDel;
+
+
+ private String resourcePath;
+ private DataStorage storage;
+
+ public JsonMetadata(String fullPath, DataStorage storage) {
+ this.resourcePath = fullPath;
+ this.storage = storage;
+ }
+
+
+ /**.
+ * Given a path, which may represent a glob pattern, a directory, or a file, this method
+ * finds the set of relevant metadata files on the storage system. The algorithm for finding the
+ * metadata file is as follows:
+ * <p>
+ * For each file represented by the path (either directly, or via a glob):
+ * If parentPath/prefix.fileName exists, use that as the metadata file.
+ * Else if parentPath/prefix exists, use that as the metadata file.
+ * <p>
+ * Resolving conflicts, merging the metadata, etc, is not handled by this method and should be
+ * taken care of by downstream code.
+ *
+ * This can go into a util package if metadata files are considered a general enough pattern
+ *
+ * @param path Path, as passed in to a LoadFunc (may be a Hadoop glob)
+ * @param prefix Metadata file designation, such as .pig_schema or .pig_stats
+ * @param storage
+ * @return Set of element descriptors for all metadata files associated with the files on the path.
+ */
+ protected Set<ElementDescriptor> findMetaFile(String path, String prefix, DataStorage storage)
+ throws IOException {
+ String fullPath = FileLocalizer.fullPath(path, storage);
+ Set<ElementDescriptor> metaFileSet = new HashSet<ElementDescriptor>();
+ if(storage.isContainer(fullPath)) {
+ ElementDescriptor metaFilePath = storage.asElement(fullPath, prefix);
+ if (metaFilePath.exists()) {
+ metaFileSet.add(metaFilePath);
+ }
+ } else {
+ ElementDescriptor[] descriptors = storage.asCollection(path);
+ for(ElementDescriptor descriptor : descriptors) {
+ String fileName = null, parentName = null;
+ ContainerDescriptor parentContainer = null;
+ if (descriptor instanceof HFile) {
+ Path descriptorPath = ((HFile) descriptor).getPath();
+ fileName = descriptorPath.getName();
+ Path parent = descriptorPath.getParent();
+ parentName = parent.toString();
+ parentContainer = new HDirectory((HDataStorage)storage,parent);
+ } else if (descriptor instanceof LocalFile) {
+ File descriptorPath = ((LocalFile) descriptor).getPath();
+ fileName = descriptorPath.getName();
+ parentName = descriptorPath.getParent();
+ parentContainer = new LocalDir((LocalDataStorage)storage,parentName);
+ }
+ ElementDescriptor metaFilePath = storage.asElement(parentName, prefix+"."+fileName);
+
+ // if the file has a custom schema, use it
+ if (metaFilePath.exists()) {
+ metaFileSet.add(metaFilePath);
+ continue;
+ }
+
+ // if no custom schema, try the parent directory
+ metaFilePath = storage.asElement(parentContainer, prefix);
+ if (metaFilePath.exists()) {
+ metaFileSet.add(metaFilePath);
+ }
+ }
+ }
+ return metaFileSet;
+ }
+
+
+ @Override
+ public String[] getPartitionKeys(String location, Configuration conf) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ /**
+ * For JsonMetadata schema is considered optional
+ * This method suppresses (and logs) errors if they are encountered.
+ *
+ * TODO location and conf params are ignored in favor of initialzation data
+ */
+ @Override
+ public ResourceSchema getSchema(String location, Configuration conf) throws IOException {
+ Set<ElementDescriptor> schemaFileSet = null;
+ try {
+ schemaFileSet = findMetaFile(resourcePath, schemaFileName, storage );
+ } catch (IOException e) {
+ log.warn("Could not find schema file for "+resourcePath);
+ return null;
+ }
+ // TODO we assume that all schemas are the same. The question of merging schemas is left open for now.
+ ElementDescriptor schemaFile = null;
+ if (!schemaFileSet.isEmpty()) {
+ schemaFile = schemaFileSet.iterator().next();
+ } else {
+ log.warn("Could not find schema file for "+resourcePath);
+ return null;
+ }
+ log.info("Found schema file: "+schemaFile.toString());
+ ResourceSchema resourceSchema = null;
+ try {
+ resourceSchema = new ObjectMapper().readValue(schemaFile.open(), ResourceSchema.class);
+ } catch (JsonParseException e) {
+ log.warn("Unable to load Resource Schema for "+resourcePath);
+ e.printStackTrace();
+ } catch (JsonMappingException e) {
+ log.warn("Unable to load Resource Schema for "+resourcePath);
+ e.printStackTrace();
+ } catch (IOException e) {
+ log.warn("Unable to load Resource Schema for "+resourcePath);
+ e.printStackTrace();
+ }
+
+ return resourceSchema;
+ }
+
+ /**
+ * For JsonMetadata stats are considered optional
+ * This method suppresses (and logs) errors if they are encountered.
+ * @see org.apache.pig.experimental.LoadMetadata#getStatistics(String, Configuration)
+ */
+ @Override
+ public ResourceStatistics getStatistics(String location, Configuration conf) throws IOException {
+ Set<ElementDescriptor> statFileSet = null;
+ try {
+ statFileSet = findMetaFile(resourcePath, statFileName, storage);
+ } catch (IOException e) {
+ log.warn("could not fine stat file for "+resourcePath);
+ return null;
+ }
+ ElementDescriptor statFile = null;
+ if (!statFileSet.isEmpty()) {
+ statFile = statFileSet.iterator().next();
+ } else {
+ log.warn("Could not find stat file for "+resourcePath);
+ return null;
+ }
+ log.info("Found stat file "+statFile.toString());
+ ResourceStatistics resourceStats = null;
+ try {
+ resourceStats = new ObjectMapper().readValue(statFile.open(), ResourceStatistics.class);
+ } catch (JsonParseException e) {
+ log.warn("Unable to load Resource Statistics for "+resourcePath);
+ e.printStackTrace();
+ } catch (JsonMappingException e) {
+ log.warn("Unable to load Resource Statistics for "+resourcePath);
+ e.printStackTrace();
+ } catch (IOException e) {
+ log.warn("Unable to load Resource Statistics for "+resourcePath);
+ e.printStackTrace();
+ }
+ return resourceStats;
+ }
+
+ @Override
+ public void setParitionFilter(OperatorPlan<Operator<PlanVisitor<?,?>>> plan)
+ throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void setStatistics(ResourceStatistics stats, String location, Configuration conf) throws IOException {
+ ElementDescriptor statFilePath = storage.asElement(resourcePath, statFileName);
+ if(!statFilePath.exists() && stats != null) {
+ try {
+ new ObjectMapper().writeValue(statFilePath.create(), stats);
+ } catch (JsonGenerationException e) {
+ log.warn("Unable to write Resource Statistics for "+resourcePath);
+ e.printStackTrace();
+ } catch (JsonMappingException e) {
+ log.warn("Unable to write Resource Statistics for "+resourcePath);
+ e.printStackTrace();
+ }
+ }
+ }
+
+
+ @Override
+ public void setSchema(ResourceSchema schema, String location, Configuration conf) throws IOException {
+ ElementDescriptor schemaFilePath = storage.asElement(resourcePath, schemaFileName);
+ if(!schemaFilePath.exists() && schema != null) {
+ try {
+ new ObjectMapper().writeValue(schemaFilePath.create(), schema);
+ } catch (JsonGenerationException e) {
+ log.warn("Unable to write Resource Statistics for "+resourcePath);
+ e.printStackTrace();
+ } catch (JsonMappingException e) {
+ log.warn("Unable to write Resource Statistics for "+resourcePath);
+ e.printStackTrace();
+ }
+ }
+ if (printHeaders) {
+ ElementDescriptor headerFilePath = storage.asElement(resourcePath, headerFileName);
+ if (!headerFilePath.exists()) {
+ OutputStream os = headerFilePath.create();
+ try {
+ String[] names = schema.fieldNames();
+
+ for (int i=0; i < names.length; i++) {
+ os.write(names[i].getBytes("UTF-8"));
+ if (i <names.length-1) {
+ os.write(fieldDel);
+ } else {
+ os.write(recordDel);
+ }
+ }
+ } finally {
+ os.close();
+ }
+ }
+ }
+ }
+
+
+ public void setFieldDel(byte fieldDel) {
+ this.fieldDel = fieldDel;
+ }
+
+
+ public void setRecordDel(byte recordDel) {
+ this.recordDel = recordDel;
+ }
+
+}
Added: hadoop/pig/trunk/src/org/apache/pig/experimental/LoadMetadata.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/experimental/LoadMetadata.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/experimental/LoadMetadata.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/experimental/LoadMetadata.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,79 @@
+/*
+ * 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.experimental;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.pig.impl.plan.Operator;
+import org.apache.pig.impl.plan.OperatorPlan;
+import org.apache.pig.impl.plan.PlanVisitor;
+
+/**
+ * This interface defines how to retrieve metadata related to data to be loaded.
+ * If a given loader does not implement this interface, it will be assumed that it
+ * is unable to provide metadata about the associated data.
+ */
+public interface LoadMetadata {
+
+ /**
+ * Get a schema for the data to be loaded.
+ * @param location Location
+ * @param conf The {@link Configuration} object
+ * @return schema for the data to be loaded. This schema should represent
+ * all tuples of the returned data. If the schema is unknown or it is
+ * not possible to return a schema that represents all returned data,
+ * then null should be returned.
+ * @throws IOException if an exception occurs while determining the schema
+ */
+ ResourceSchema getSchema(String location, Configuration conf) throws IOException;
+
+ /**
+ * Get statistics about the data to be loaded. If no statistics are
+ * available, then null should be returned.
+ * @param location Location
+ * @param conf The {@link Configuration} object
+ * @return statistics about the data to be loaded. If no statistics are
+ * available, then null should be returned.
+ * @throws IOException if an exception occurs while retrieving statistics
+ */
+ ResourceStatistics getStatistics(String location, Configuration conf)
+ throws IOException;
+
+ /**
+ * Find what columns are partition keys for this input.
+ * @param location Location
+ * @param conf The {@link Configuration} object
+ * @return array of field names of the partition keys.
+ * @throws IOException if an exception occurs while retrieving partition keys
+ */
+ String[] getPartitionKeys(String location, Configuration conf)
+ throws IOException;
+
+ /**
+ * Set the filter for partitioning. It is assumed that this filter
+ * will only contain references to fields given as partition keys in
+ * getPartitionKeys
+ * @param plan that describes filter for partitioning
+ * @throws IOException if the filter is not compatible with the storage
+ * mechanism or contains non-partition fields.
+ */
+ void setParitionFilter(OperatorPlan<Operator<PlanVisitor<?,?>>> plan) throws IOException;
+
+}
Added: hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceSchema.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceSchema.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceSchema.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceSchema.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,182 @@
+/*
+ * 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.experimental;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+
+public class ResourceSchema {
+
+ /* Array Getters intentionally return mutable arrays instead of copies,
+ * to simplify updates without unnecessary copying.
+ * Setters make a copy of the arrays in order to prevent an array
+ * from being shared by two objects, with modifications in one
+ * accidentally changing the other.
+ */
+
+ // initializing arrays to empty so we don't have to worry about NPEs
+ // setters won't set to null
+ public ResourceFieldSchema[] fields = new ResourceFieldSchema[0];
+ public Map<String, Integer> byName = new HashMap<String, Integer>();
+
+ public enum Order { ASCENDING, DESCENDING }
+ public int[] sortKeys = {}; // each entry is an offset into the fields array.
+ public Order[] sortKeyOrders = new Order[0];
+
+
+ public int version = 0;
+
+ public static class ResourceFieldSchema {
+ public String name;
+
+ // values are constants from DataType
+ public byte type;
+
+ public String description;
+
+ // nested tuples and bags will have their own schema
+ public ResourceSchema schema;
+
+ public ResourceFieldSchema() {
+
+ }
+
+ public ResourceFieldSchema(FieldSchema fieldSchema) {
+ type = fieldSchema.type;
+ name = fieldSchema.alias;
+ description = "autogenerated from Pig Field Schema";
+ if (type == DataType.BAG || type == DataType.TUPLE) {
+ schema = new ResourceSchema(fieldSchema.schema);
+ } else {
+ schema = null;
+ }
+ }
+
+ public String getName() {
+ return name;
+ }
+ public ResourceFieldSchema setName(String name) {
+ this.name = name;
+ return this;
+ }
+
+ public byte getType() {
+ return type;
+ }
+ public ResourceFieldSchema setType(byte type) {
+ this.type = type;
+ return this;
+ }
+
+ public String getDescription() {
+ return description;
+ }
+
+ public ResourceFieldSchema setDescription(String description) {
+ this.description = description;
+ return this;
+ }
+
+ public ResourceSchema getSchema() {
+ return schema;
+ }
+
+ public ResourceFieldSchema setSchema(ResourceSchema schema) {
+ this.schema = schema;
+ return this;
+ }
+ }
+
+
+ public ResourceSchema() {
+
+ }
+
+ public ResourceSchema(Schema pigSchema) {
+ List<FieldSchema> pigSchemaFields = pigSchema.getFields();
+ fields = new ResourceFieldSchema[pigSchemaFields.size()];
+ byName = new HashMap<String, Integer>(fields.length);
+ for (int i=0; i<fields.length; i++) {
+ fields[i] = new ResourceFieldSchema(pigSchemaFields.get(i));
+ byName.put(fields[i].getName(), i);
+ }
+
+ }
+
+ public int getVersion() {
+ return version;
+ }
+
+ public ResourceSchema setVersion(int version) {
+ this.version = version;
+ return this;
+ }
+
+ public ResourceFieldSchema[] getFields() {
+ return fields;
+ }
+
+ public String[] fieldNames() {
+ String[] names = new String[fields.length];
+ for (int i=0; i<fields.length; i++) {
+ names[i] = fields[i].getName();
+ }
+ return names;
+ }
+
+ public ResourceSchema setFields(ResourceFieldSchema[] fields) {
+ if (fields != null)
+ this.fields = Arrays.copyOf(fields, fields.length);
+ return this;
+ }
+
+ public Map<String, Integer> getByName() {
+ return byName;
+ }
+
+ public ResourceSchema setByName(Map<String, Integer> byName) {
+ this.byName = byName;
+ return this;
+ }
+
+ public int[] getSortKeys() {
+ return sortKeys;
+ }
+ public ResourceSchema setSortKeys(int[] sortKeys) {
+ if (sortKeys != null)
+ this.sortKeys = Arrays.copyOf(sortKeys, sortKeys.length);
+ return this;
+ }
+
+ public Order[] getSortKeyOrders() {
+ return sortKeyOrders;
+ }
+
+ public ResourceSchema setSortKeyOrders(Order[] sortKeyOrders) {
+ if (sortKeyOrders != null)
+ this.sortKeyOrders = Arrays.copyOf(sortKeyOrders, sortKeyOrders.length);
+ return this;
+ }
+}
\ No newline at end of file
Added: hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceStatistics.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceStatistics.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceStatistics.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/experimental/ResourceStatistics.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,259 @@
+/*
+ * 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.experimental;
+
+import java.io.Serializable;
+import java.util.Arrays;
+
+public class ResourceStatistics implements Serializable, Cloneable {
+
+ /* Getters intentionally return mutable arrays instead of copies,
+ * to simplify updates without unnecessary copying.
+ * Setters make a copy of the arrays in order to prevent an array
+ * from being shared by two objects, with modifications in one
+ * accidentally changing the other.
+ */
+
+ // arrays are initialized to empty so we don't have to worry about NPEs
+ // setters disallow setting them to null.
+
+ private static final long serialVersionUID = 1L;
+ public Long mBytes; // size in megabytes
+ public Long numRecords; // number of records
+ public Long avgRecordSize;
+ public ResourceFieldStatistics[] fields = new ResourceFieldStatistics[0];
+
+ public static class ResourceFieldStatistics implements Serializable {
+
+ public static final long serialVersionUID = 1L;
+
+ public int version;
+
+ public Long numDistinctValues; // number of distinct values represented in this field
+
+ // We need some way to represent a histogram of values in the field,
+ // as those will be useful. However, we can't count on being
+ // able to hold such histograms in memory. Have to figure out
+ // how they can be kept on disk and represented here.
+
+ // for now.. don't create so many buckets you can't hold them in memory
+
+ // an ordered array of the most common values,
+ // in descending order of frequency
+ public Object[] mostCommonValues = new Object[0];
+
+ // an array that matches the mostCommonValues array, and lists
+ // the frequencies of those values as a fraction (0 through 1) of
+ // the total number of records
+ public float[] mostCommonValuesFreq = new float[0];
+
+ // an ordered array of values, from min val to max val
+ // such that the number of records with values
+ // between valueHistogram[i] and and valueHistogram[i+1] is
+ // roughly equal for all values of i.
+ // NOTE: if mostCommonValues is non-empty, the values in that array
+ // should not be included in the histogram. Adjust accordingly.
+ public Object[] valueHistogram = new Object[0];
+
+
+ public int getVersion() {
+ return version;
+ }
+
+ public ResourceFieldStatistics setVersion(int version) {
+ this.version = version;
+ return this;
+ }
+
+ public Long getNumDistinctValues() {
+ return numDistinctValues;
+ }
+
+ public ResourceFieldStatistics setNumDistinctValues(Long numDistinctValues) {
+ this.numDistinctValues = numDistinctValues;
+ return this;
+ }
+
+ public Object[] getMostCommonValues() {
+ return mostCommonValues;
+ }
+
+ public ResourceFieldStatistics setMostCommonValues(Object[] mostCommonValues) {
+ if (mostCommonValues !=null)
+ this.mostCommonValues =
+ Arrays.copyOf(mostCommonValues, mostCommonValues.length);
+ return this;
+ }
+
+ public float[] getMostCommonValuesFreq() {
+ return mostCommonValuesFreq;
+ }
+
+ public ResourceFieldStatistics setMostCommonValuesFreq(float[] mostCommonValuesFreq) {
+ if (mostCommonValuesFreq != null)
+ this.mostCommonValuesFreq =
+ Arrays.copyOf(mostCommonValuesFreq, mostCommonValuesFreq.length);
+ return this;
+ }
+
+ public Object[] getValueHistogram() {
+ return valueHistogram;
+ }
+
+ public ResourceFieldStatistics setValueHistogram(Object[] valueHistogram) {
+ if (valueHistogram != null)
+ this.valueHistogram = Arrays.copyOf(valueHistogram, valueHistogram.length);
+ return this;
+ }
+
+
+ /*
+ * equals() and hashCode() overridden mostly for ease of testing
+ * you shouldn't encounter a situation in which you need to .equals()
+ * two sets of statistics on different objects "in the wild"
+ */
+ @Override
+ public boolean equals(Object anOther) {
+ if (anOther == null || !(anOther.getClass().equals(this.getClass())))
+ return false;
+ ResourceFieldStatistics other = (ResourceFieldStatistics) anOther;
+ // setters do not allow null values, so no worries about NPEs here
+ return (Arrays.equals(mostCommonValues, other.mostCommonValues) &&
+ Arrays.equals(mostCommonValuesFreq, other.mostCommonValuesFreq) &&
+ Arrays.equals(valueHistogram, other.valueHistogram) &&
+ this.numDistinctValues.equals(other.numDistinctValues) &&
+ this.version == other.version
+ );
+ }
+
+ /**
+ * A naive hashCode implementation following the example in IBM's developerworks:
+ * http://www.ibm.com/developerworks/java/library/j-jtp05273.html
+ */
+ @Override
+ public int hashCode() {
+ int hash = 1;
+ hash = 31 * hash + Arrays.hashCode(mostCommonValues);
+ hash = 31 * hash + Arrays.hashCode(mostCommonValuesFreq);
+ hash = 31 * hash + numDistinctValues.hashCode();
+ hash = 31 * hash + Arrays.hashCode(valueHistogram);
+ hash = 31 * hash + version;
+ return 0;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("ResourceStatistics. Version: "+version+"\n");
+ sb.append("MCV:\n");
+ for (Object o : mostCommonValues) sb.append('['+ o.toString() +']');
+ sb.append("\n MCVfreq:\n");
+ for (Float f : mostCommonValuesFreq) sb.append('['+f.toString()+']');
+ sb.append("\n");
+ sb.append("numDistVals: "+numDistinctValues);
+ sb.append("valHistogram: \n");
+ for (Object o : valueHistogram) sb.append('['+o.toString()+']');
+ sb.append("\n");
+ return sb.toString();
+ }
+ }
+
+
+ public Long getmBytes() {
+ return mBytes;
+ }
+ public ResourceStatistics setmBytes(Long mBytes) {
+ this.mBytes = mBytes;
+ return this;
+ }
+ public Long getNumRecords() {
+ return numRecords;
+ }
+ public ResourceStatistics setNumRecords(Long numRecords) {
+ this.numRecords = numRecords;
+ return this;
+ }
+
+ /*
+ * returns average record size. This number can be explicitly specified by statistics, or
+ * if absent, computed using totalbytes/totalrecords. Will return null if can't be computed.
+ */
+ public Long getAvgRecordSize() {
+ if (avgRecordSize == null && (mBytes != null && numRecords != null))
+ return mBytes / numRecords;
+ else
+ return avgRecordSize;
+ }
+
+ public void setAvgRecordSize(Long size) {
+ avgRecordSize = size;
+ }
+
+ public ResourceFieldStatistics[] getFields() {
+ return fields;
+ }
+
+ public ResourceStatistics setFields(ResourceFieldStatistics[] fields) {
+ if (fields != null)
+ this.fields = Arrays.copyOf(fields, fields.length);
+ return this;
+ }
+
+
+ /*
+ * equals() and hashCode() overridden mostly for ease of testing
+ * you shouldn't encounter a situation in which you need to .equals()
+ * two sets of statistics on different objects "in the wild"
+ */
+ @Override
+ public boolean equals(Object anOther) {
+ if (anOther == null || !(anOther.getClass().equals(this.getClass())))
+ return false;
+ ResourceStatistics other = (ResourceStatistics) anOther;
+ return (Arrays.equals(fields, other.fields) &&
+ ((mBytes==null)
+ ? (other.mBytes==null) : mBytes.equals(other.mBytes)) &&
+ ((numRecords == null)
+ ? (other.numRecords==null) : numRecords.equals(other.numRecords))
+ );
+ }
+
+ @Override
+ public int hashCode() {
+ int hash = 1;
+ hash = 31*hash + Arrays.hashCode(fields);
+ hash = 31*hash + (mBytes == null ? 0 : mBytes.hashCode());
+ hash = 31*hash + (numRecords == null ? 0 : numRecords.hashCode());
+ return hash;
+ }
+ // Probably more in here
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("Field Stats: \n");
+ for (ResourceFieldStatistics f : fields) sb.append(f.toString());
+ sb.append("mBytes: "+mBytes);
+ sb.append("numRecords: "+numRecords);
+ return sb.toString();
+ }
+
+ public Object clone() throws CloneNotSupportedException {
+ return super.clone();
+ }
+
+}
\ No newline at end of file
Added: hadoop/pig/trunk/src/org/apache/pig/experimental/StoreMetadata.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/experimental/StoreMetadata.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/experimental/StoreMetadata.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/experimental/StoreMetadata.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,44 @@
+/*
+ * 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.experimental;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+
+/**
+ * This interface defines how to write metadata related to data to be loaded.
+ * If a given store function does not implement this interface, it will be assumed that it
+ * is unable to record metadata about the associated data.
+ */
+
+public interface StoreMetadata {
+
+ /**
+ * Set statistics about the data being written.
+ * @throws IOException
+ */
+ void setStatistics(ResourceStatistics stats, String location, Configuration conf) throws IOException;
+
+ /**
+ * Set schema of the data being written
+ * @throws IOException
+ */
+ void setSchema(ResourceSchema schema, String location, Configuration conf) throws IOException;
+}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java?rev=891419&r1=891418&r2=891419&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java Wed Dec 16 20:37:14 2009
@@ -314,7 +314,7 @@
}
}
- private static String fullPath(String fileName, DataStorage storage) {
+ public static String fullPath(String fileName, DataStorage storage) {
String fullPath;
try {
if (fileName.charAt(0) != '/') {
Added: hadoop/pig/trunk/test/org/apache/pig/test/TestResourceSchema.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestResourceSchema.java?rev=891419&view=auto
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestResourceSchema.java (added)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestResourceSchema.java Wed Dec 16 20:37:14 2009
@@ -0,0 +1,73 @@
+/*
+ * 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 junit.framework.TestCase;
+
+import org.apache.pig.experimental.ResourceSchema;
+import org.apache.pig.backend.executionengine.ExecException;
+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.SchemaMergeException;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+import org.apache.pig.test.utils.TypeCheckingTestUtil;
+import org.junit.Test;
+
+import junit.framework.TestCase;
+
+public class TestResourceSchema extends TestCase {
+
+ /**
+ * Test that ResourceSchema is correctly created given a
+ * pig.Schema and vice versa
+ * @throws FrontendException
+ * @throws SchemaMergeException
+ * @throws ExecException
+ */
+ @Test
+ public void testResourceFlatSchemaCreation() throws ExecException, SchemaMergeException, FrontendException {
+ String [] aliases ={"f1", "f2"};
+ byte[] types = {DataType.CHARARRAY, DataType.INTEGER};
+ Schema origSchema = TypeCheckingTestUtil.genFlatSchema(
+ aliases,types);
+ ResourceSchema rsSchema = new ResourceSchema(origSchema);
+ assertEquals("num fields", aliases.length, rsSchema.getFields().length);
+ ResourceSchema.ResourceFieldSchema[] fields = rsSchema.getFields();
+ for (int i=0; i<fields.length; i++) {
+ assertEquals(fields[i].getName(), aliases[i]);
+ assertEquals(fields[i].getType(), types[i]);
+ }
+ Schema genSchema = new Schema();
+ for (ResourceSchema.ResourceFieldSchema field : rsSchema.getFields()) {
+ FieldSchema pigFieldSchema;
+ pigFieldSchema = DataType.determineFieldSchema(field);
+
+ // determineFieldSchema only sets the types. we also want the aliases.
+ // TODO this doesn't work properly for complex types
+ // (hence the "true" for relaxInner below)
+ pigFieldSchema.alias = field.getName();
+
+ genSchema.add(pigFieldSchema);
+ }
+ assertTrue("generated schema equals original" , Schema.equals(genSchema, origSchema, true, false));
+ }
+
+
+}