Pig needs to know which InputFormat to use for reading your input. It calls getInputFormat to get an instance of the input format. It gets an instance rather than the class itself so that your load function can control the instantiation: any generic parameters, constructor arguments, etc. For our example load function, this method is very simple. It uses TextInputFormat, an input format that reads text data from HDFS files.

// JsonLoader.java
public InputFormat getInputFormat() throws IOException {
    return new TextInputFormat();
}

Pig communicates the location string provided by the user to the load function via setLocation. So if the load operator in Pig Latin is A = load 'input'; then “input” is the location string. This method is called on both the front and back ends, possibly multiple times. Thus you need to take care that this method does not do anything that will cause problems if done more than one time. Your load function should communicate the location to its input format. For example, JsonLoader passes the filename via a helper method on FileInputFormat (a super class of TextInputFormat).

// JsonLoader.java
public void setLocation(String location, Job job) throws IOException {
    FileInputFormat.setInputPaths(job, location);
}

The Hadoop Job is passed along with the location, since that is the location where input formats usually store their configuration information.

setLocation is called on both the front and back ends because input formats store their location in the Job object, as shown in the above example. For MapReduce jobs, that always have only one input, this works. For Pig jobs, where the same input format may be used to load multiple different inputs, such as in the join or union case, one instance of the input path will overwrite another in the Job object. To work around this, Pig remembers the location in an input specific parameter, and calls setLocation again on the backend so that the input format can get itself set up properly before reading.

For files on HDFS the location provided by the user may be a relative location, rather than absolute. To deal with this Pig needs to resolve these to absolute locations based on the current working directory at the time of the load. Consider the following Pig Latin:

cd /user/joe;
input1 = load 'input';
cd /user/fred;
input2 = load 'input';

These two load statements should load different files. But Pig cannot assume it understands how to turn a relative path into an absolute path, because it does not know what that input is. It could be an HDFS path, a database table name, etc. So it leaves this to the load function. Before calling setLocation Pig passes the location string to relativeToAbsolutePath to do any necessary conversion. Since most loaders are reading from HDFS, the default implementation in LoadFunc handles the HDFS case. If your loading will never need to do this conversion, it should override this method and return the location string passed to it.

Some Pig functions, such as PigStorage and HBaseStorage, load data by default without understanding its type information. They place the data unchanged in DataByteArray objects. At a later time when Pig needs to cast that data to another type, it does not know how to, since it does not understand how the data is represented in the bytearray. Therefore it relies on the load function to provide a method to cast from bytearray to the appropriate type.

Pig determines which set of casting functions to use by calling getLoadCaster on the load function. This should return either null, which indicates that your load function does not expect to do any bytearray casts, or an implementation of the LoadCaster interface, which will be used to do the casts. We will look at the methods of LoadCaster in the section called “Casting Bytearrays” below.

Our example loader returns null, because it provides typed data based on the stored schema, and therefore does not expect to be casting data. Any bytearrays in its data are binary data that should not be cast.

Before reading any data, Pig gives your load function a chance to set itself up by calling prepareToRead. This is called in each map task. It passes a copy of the RecordReader, which your load function will need later to read records from the input. RecordReader is a class that InputFormat uses to read records from an input split. Pig obtains the record reader it passes to prepareToRead by calling getRecordReader on the input format that your store function returned from getInputFormat. It also passes an instance of the PigSplit which contains the Hadoop InputSplit corresponding to the partition of input this instance of your load function will read. If you need split-specific information you can get it from here.

Our example loader, beyond storing the record reader, also reads the schema file that was stored into UDFContext in the front end so that it knows how to parse the input file. Notice how it uses the signature passed in setUDFContextSignature to access the appropriate properties object. Finally, it creates a JsonFactory object that is used to generate a parser for each line.

// JsonLoader.java
public void prepareToRead(RecordReader reader, PigSplit split)
throws IOException {
    this.reader = reader;
        
    // Get the schema string from the UDFContext object.
    UDFContext udfc = UDFContext.getUDFContext();
    Properties p =
        udfc.getUDFProperties(this.getClass(), new String[]{udfcSignature});
    String strSchema = p.getProperty("pig.jsonloader.schema");
    if (strSchema == null) {
        throw new IOException("Could not find schema in UDF context");
    }

    // Parse the schema from the string stored in the properties object.
    ResourceSchema schema =
        new ResourceSchema(Utils.getSchemaFromString(strSchema));
    fields = schema.getFields();

    jsonFactory = new JsonFactory();
}

Now we are to the meat of your load function, reading records from its record reader and returning tuples to Pig. Pig will call getNext and place the resulting tuple into its processing pipeline. It will keep doing this until getNext returns a null, which indicates that the input for this split has been fully read.

Pig does not copy the tuple that results from this method, but feeds it directly to its pipeline to avoid the copy overhead. This means this method cannot re-use objects, but must create a new tuple and contents for each record it reads. On the other hand, record readers may choose to re-use their key and value objects from record to record; most standard implementations do. So, before writing a loader that tries to be efficient and wrap the keys and values from the record reader directly into the tuple to avoid a copy, you must make sure you understand how the record reader is managing its data.

For information on creating the appropriate Java objects when constructing tuples for Pig, see the section called “Interacting with Pig Values”.

Our sample load function's implementation of getNext reads the value from the Hadoop record (the key is ignored), constructs a JsonParser to parse it, parses the fields, and returns the resulting tuple. If there are parse errors it does not throw an exception. Instead it returns a tuple with null fields where the data could not be parsed. This prevents bad lines from causing the whole job to fail. Warnings are issued so that users can see which records were ignored.

// JsonLoader.java
public Tuple getNext() throws IOException {
    Text val = null;
    try {
        // Read the next key value pair from the record reader.  If it's
        // finished, return null
        if (!reader.nextKeyValue()) return null;

        // Get the current value.  We don't use the key.
        val = (Text)reader.getCurrentValue();
    } catch (InterruptedException ie) {
        throw new IOException(ie);
    }
    // Create a parser specific for this input line.  This may not be the
    // most efficient approach.
    ByteArrayInputStream bais = new ByteArrayInputStream(val.getBytes());
    JsonParser p = jsonFactory.createJsonParser(bais);

    // Create the tuple we will be returning.  We create it with the right
    // number of fields, as the Tuple object is optimized for this case.
    Tuple t = tupleFactory.newTuple(fields.length);

    // Read the start object marker.  Throughout this file if the parsing
    // isn't what we expect we return a tuple with null fields rather than
    // throwing an exception.  That way a few mangled lines don't fail the job.
    if (p.nextToken() != JsonToken.START_OBJECT) {
        log.warn("Bad record, could not find start of record " + val.toString());
        return t;
    }

    // Read each field in the record
    for (int i = 0; i < fields.length; i++) {
        t.set(i, readField(p, fields[i], i));
    }

    if (p.nextToken() != JsonToken.END_OBJECT) {
        log.warn("Bad record, could not find end of record " +
            val.toString());
        return t;
    }
    p.close();
    return t;
}

private Object readField(JsonParser p,
                         ResourceFieldSchema field,
                         int fieldnum) throws IOException {
    // Read the next token
    JsonToken tok = p.nextToken();
    if (tok == null) {
        log.warn("Early termination of record, expected " + fields.length
            + " fields bug found " + fieldnum);
        return null;
    }   

    // Check to see if this value was null
    if (tok == JsonToken.VALUE_NULL) return null;

    // Read based on our expected type
    switch (field.getType()) {
    case DataType.INTEGER:
        // Read the field name
        p.nextToken();
        return p.getValueAsInt();

    case DataType.LONG:
        p.nextToken();
        return p.getValueAsLong();

    case DataType.FLOAT: 
        p.nextToken();
        return (float)p.getValueAsDouble();

    case DataType.DOUBLE: 
        p.nextToken();
        return p.getValueAsDouble();

    case DataType.BYTEARRAY:
        p.nextToken();
        byte[] b = p.getBinaryValue();
        // Use the DBA constructor that copies the bytes so that we own
        // the memory
        return new DataByteArray(b, 0, b.length);

    case DataType.CHARARRAY:
        p.nextToken();
        return p.getText();

    case DataType.MAP:
        // Should be a start of the map object
        if (p.nextToken() != JsonToken.START_OBJECT) {
            log.warn("Bad map field, could not find start of object, field "
                + fieldnum);
            return null;
        }
        Map<String, String> m = new HashMap<String, String>();
        while (p.nextToken() != JsonToken.END_OBJECT) {
            String k = p.getCurrentName();
            String v = p.getText();
            m.put(k, v);
        }
        return m;

    case DataType.TUPLE:
        if (p.nextToken() != JsonToken.START_OBJECT) {
            log.warn("Bad tuple field, could not find start of object, "
                + "field " + fieldnum);
            return null;
        }

        ResourceSchema s = field.getSchema();
        ResourceFieldSchema[] fs = s.getFields();
        Tuple t = tupleFactory.newTuple(fs.length);

        for (int j = 0; j < fs.length; j++) {
            t.set(j, readField(p, fs[j], j));
        }

        if (p.nextToken() != JsonToken.END_OBJECT) {
            log.warn("Bad tuple field, could not find end of object, "
                + "field " + fieldnum);
            return null;
        }
        return t;

    case DataType.BAG:
        if (p.nextToken() != JsonToken.START_ARRAY) {
            log.warn("Bad bag field, could not find start of array, "
                + "field " + fieldnum);
            return null;
        }

        s = field.getSchema();
        fs = s.getFields();
        // Drill down the next level to the tuple's schema.
        s = fs[0].getSchema();
        fs = s.getFields();

        DataBag bag = bagFactory.newDefaultBag();

        JsonToken innerTok;
        while ((innerTok = p.nextToken()) != JsonToken.END_ARRAY) {
            if (innerTok != JsonToken.START_OBJECT) {
                log.warn("Bad bag tuple field, could not find start of "
                    + "object, field " + fieldnum);
                return null;
            }

            t = tupleFactory.newTuple(fs.length);
            for (int j = 0; j < fs.length; j++) {
                t.set(j, readField(p, fs[j], j));
            }

            if (p.nextToken() != JsonToken.END_OBJECT) {
                log.warn("Bad bag tuple field, could not find end of "
                    + "object, field " + fieldnum);
                return null;
            }
            bag.add(t);
        }
        return bag;

    default:
        throw new IOException("Unknown type in input schema: " +
            field.getType());
    }

}

Many data storage mechanisms can record the schema along with the data. Pig does not assume the ability to store schemas. But if your storage can store the schema, it can be very useful. This frees script writers from needing to specify the field names and types as part of the load operator in Pig Latin. This is user friendly, less error prone, and avoids the need to rewrite scripts when the schema of your data changes.

Some types of data storage also partition the data. If Pig understands this partitioning, it can load only those partitions that are needed for a particular script. Both of these functions are enabled by implementing the LoadMetadata interface.

getSchema in the LoadMetadata interface gives your load function a chance to provide a schema. It is passed the location string provided by the user as well as the Hadoop Job object, in case it needs information in this object to open the schema. It is expected to return a ResourceSchema that represents the data that will be returned. ResourceSchema is very similar to the Schema class used by evaluation functions. See the section called “Input and Output Schemas” for details. There is one important difference. In ResourceFieldSchema, the schema object associated with a bag always has one field, which is a tuple. The schema for the tuples in the bag is described by that tuple's ResourceFieldSchema.

Our example load and store functions store the schema in a side file^[27] named _schema in HDFS. Our implementation of getSchema reads this file. It also serializes the schema into UDFContext so that it is available on the back end.

// JsonLoader.java
public ResourceSchema getSchema(String location, Job job)
throws IOException {
    // Open the schema file and read the schema
    // Get an HDFS handle.
    FileSystem fs = FileSystem.get(job.getConfiguration());
    DataInputStream in = fs.open(new Path(location + "/_schema"));
    String line = in.readLine();
    in.close();

    // Parse the schema
    ResourceSchema s = new ResourceSchema(Utils.getSchemaFromString(line));
    if (s == null) {
        throw new IOException("Unable to parse schema found in file " +
            location + "/_schema");
    }

    // Now that we have determined the schema, store it in our
    // UDFContext properties object so we have it when we need it on the
    // backend
    UDFContext udfc = UDFContext.getUDFContext();
    Properties p =
        udfc.getUDFProperties(this.getClass(), new String[]{udfcSignature});
    p.setProperty("pig.jsonloader.schema", line);

    return s;
}

Once your loader implements getSchema, load statements that use your loader do not need to declare their schemas in order for the field names to be used in the script. For example, if we had data with a schema of user:chararray, age:int, gpa:double, the following Pig Latin will compile and run:

register 'acme.jar';
register 'src/pig/trunk/build/ivy/lib/Pig/jackson-core-asl-1.6.0.jar';
      
A = load 'input' using com.acme.io.JsonLoader();
B = foreach A generate user;
dump B;

LoadMetadata also includes a getStatistics method. Pig does not yet make use of statistics in job planning; this method is for future use.

Some types of storage partition their data, allowing you to read only the relevant sections for a given job. The LoadMetadata interface also provides methods for working with partitions in your data. In order for Pig to request the relevant partitions, it must know how the data is partitioned. Pig determines this by calling getPartitionKeys. If this returns a null, or the LoadMetadata interface is not implemented by your loader, then Pig will assume it needs to read the entire input.

Pig expects getPartitionKeys to return an array of strings, where each string represents one field name. Those fields are the keys used to partition the data. Pig will look for a filter statement immediately following the load statement that includes one or more of these fields. If such a statement is found, it will be passed to setPartitionFilter. If the filter includes both partition and non-partition keys and it is can be split^[28] Pig will split it and pass just the partition key-related expression to setPartitionFilter. As an example, consider an HCatalog^[29] table web_server_logs that is partitioned by two fields, date and colo.

logs    = load 'web_server_logs' using HCatLoader();
cleaned = filter logs by date = '20110614' and NotABot(user_id);
...  

Pig will call getPartitionKeys and HCatLoader will return two key names, date and colo. Pig will find the date field in the filter statement and rewrite the as shown below, pushing down the date = '20110614' predicate to HCatLoader via setPartitionFilter

logs    = load 'web_server_logs' using HCatLoader();
cleaned = filter logs by NotABot(user_id);
...  

. It is now up to HCatalog loader to assure that it only returns data from web_server_logs where date is 20110614.

The one exception to this is fields used in eval funcs or filter funcs. Pig assumes that loaders do not understand how to invoke UDFs, so Pig will not push these expressions.

Our example loader does not implement getPartitionKeys or setPartitionFilter, since it works on file data. For an example implementation of these methods, see the HCatalog code at http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatLoader.java?view=markup.

If you need to control how binary data that your loader loads is cast to other data types you can implement the LoadCaster interface. Since this interface contains a lot of methods, implementers often implement it as a separate class. This also allows load functions to share implementations of LoadCaster, since Java does not support multiple inheritance.

The interface consists of a series of methods: bytesToInteger, bytesToLong, etc. These will be called to convert a bytearray to the appropriate type. Starting in 0.9, there are two bytesToMap methods. You should implement the one that takes a ResourceFieldSchema. The other one is for backward compatibility. The bytesToBag, bytesToTuple, and bytesToMap methods take a ResourceFieldSchema that describes the field being converted. Calling getSchema on this object will return a schema that describes this bag, tuple, or map, if one exists. If Pig does not know the intended structure of the object, getSchema will return null. Keep in mind that the schema of the bag will be one field, a tuple, which in turn will have a schema describing the contents of that tuple.

A default load caster Utf8StorageConverter is provided. It handles converting UTF8 encoded text to Pig types. Scalar conversions are done in a straight forward way. Maps are expected to be surrounded by “[]” (square brackets), have keys separated by values with “#” (hash) and key value pairs separated by “,” (commas). Tuples are surrounded by “()” (parenthesis) and have fields separated by “,” (commas). Bags are surrounded by “{}” (braces) and have tuples separated by “,” (commas). There is no ability to escape these special characters.

Often a Pig Latin script will only need to read a few of the fields in the input. Some types of storage formats store their data by fields instead of by records (for example Hive's RCFile). For these types of formats there is a significant performance gain to be had by only loading fields that will be used in the script. Even for record oriented storage formats it can be useful to skip de-serializing fields that will not be used.

As part of its optimizations Pig analyzes Pig Latin scripts and determines what fields in an input it needs at each step in the script. It uses this information to aggressively drop fields it no longer needs. If the loader implements the LoadPushDown interface, Pig can go a step further and provide this information to the loader.

Once Pig knows the fields it needs, it assembles them in a RequiredFieldList and passes that to pushProjection. The load function can reply whether it can meet the request or not. It responds with a RequiredFieldResponse, which is a fancy wrapper around a Boolean. If the Boolean is true, then Pig will assume that only the required fields are being returned from getNext. If it is false, then Pig will assume that all fields are being returned by getNext and it will handle dropping the extra ones itself.

The RequiredField class used to describe which fields are required is slightly complex. Beyond allowing a user to specify whether a given field is required, it provides the ability to specify which sub-fields of that field are required. For example, for maps certain keys can be listed as required. For tuples and bags, certain fields can be listed as required.

Load functions that implement LoadPushDown should not modify the schema object returned by getSchema. This should always be the schema of the full input. Pig will manage the translation between the schema having all of the fields and the results of getNext having only some.

Our example loader does not implement LoadPushDown. For an example of a loader that does, see the HCatLoader at http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatLoader.java?view=markup.

Pig calls getOutputFormat to get an instance of the output format that your store function will use to store records. This method returns an instance rather than the classname or the class itself. This allows your store function to control how the class is instantiated. Our example store function, JsonStorage uses TextOutputFormat. This is an output format that stores text data in HDFS. We have to instantiate this with a key of LongWritable and a value of Text to match the expectations of TextInputFormat.

// JsonStorage.java
public OutputFormat getOutputFormat() throws IOException {
    return new TextOutputFormat<LongWritable, Text>();
}

Pig calls setStoreLocation to communicate the location string provided by the user to your store function. Given the Pig Latin store Z into 'output';, “output” is the location string. This method is called on both the front end and the back end and could be called multiple times. For this reason it should not have any side effects that will cause a problem if they happen multiple times. Your store function will need to communicate the location to its output format. Our example store function uses the FileOutputFormat utility function setOutputPath to do this.

// JsonStorage.java
public void setStoreLocation(String location, Job job) throws IOException {
    FileOutputFormat.setOutputPath(job, new Path(location));
}

The Hadoop Job is passed to this function as well. Most output formats store the location information in the job.

Pig calls setStoreLocation on both the front and back end because output formats usually store their location in the job, as we see in our example store function. This works for MapReduce jobs, where a single output format is guaranteed. But due to the split operator, Pig can have more than one instance of the same store function in a job. If multiple instances of a store function call FileOutputFormat.setOutputPath whichever instance calls it last will overwrite the others. Pig avoids this by keeping output-specific information and calling setStoreLocation again on the backend so that it can properly configure the output format.

For HDFS files the user may provide a relative path. Pig needs to resolve these to absolute paths using the current working directory at the time the store is called. To accomplish this Pig calls relToAbsPathForStoreLocation with the user-provided location string before calling setStoreLocation. This method translates between relative and absolute paths. For store functions writing to HDFS, the default implementation in StoreFunc handles the conversion. If you are writing a store function that does not use file paths, e.g. HBase, you should override this method to return the string it is passed.

As part of front end planning Pig gives your store function a chance to check the schema of the data to be stored. If you are storing data to a system that expects a certain schema for the output, such as an RDBMS, or you cannot store certain data types, this is the place to perform those checks. Oddly enough this method returns a void rather than a Boolean. So if you detect an issue with the schema you must throw an IOException.

Our example store function does not have limitations on the schemas it can store. However, it uses this function as a place to serialize the schema into UDFContext so that it can be used on the back end when writing data.

// JsonStorage.java

public void checkSchema(ResourceSchema s) throws IOException {
    UDFContext udfc = UDFContext.getUDFContext();
    Properties p = 
        udfc.getUDFProperties(this.getClass(), new String[]{udfcSignature});
    p.setProperty("pig.jsonstorage.schema", s.toString());
}

Pig calls your store function's prepareToWrite method in each map or reduce task before writing any data. This call passes a RecordWriter instance to use when writing data. RecordWriter is a class that OutputFormat uses to write individual records. Pig will get the record writer it passes to your store function by calling getRecordWriter on the output format your store function returned from getOutputFormat. Your store function will need to keep this reference so that it can be used in putNext.

The example store function JsonStorage also uses this method to read the schema out of the UDFContext. It will use this schema when storing data. Finally, it creates a JsonFactory for use in putNext.

// JsonStorage.java
public void prepareToWrite(RecordWriter writer) throws IOException {
    // Store the record writer reference so we can use it when it's time
    // to write tuples
    this.writer = writer;

    // Get the schema string from the UDFContext object.
    UDFContext udfc = UDFContext.getUDFContext();
    Properties p =
        udfc.getUDFProperties(this.getClass(), new String[]{udfcSignature});
    String strSchema = p.getProperty("pig.jsonstorage.schema");
    if (strSchema == null) {
        throw new IOException("Could not find schema in UDF context");
    }

    // Parse the schema from the string stored in the properties object.
    ResourceSchema schema =
        new ResourceSchema(Utils.getSchemaFromString(strSchema));
    fields = schema.getFields();

    // Build a Json factory
    jsonFactory = new JsonFactory();
    jsonFactory.configure(
        JsonGenerator.Feature.WRITE_NUMBERS_AS_STRINGS, false);
}

putNext is the core method in the store function class. Pig calls this method for every tuple it needs to store. Your store function needs to take these tuples and produce the key value pairs that its output format expects. For information on the Java objects that the data will be stored in and how to extract them see the section called “Interacting with Pig Values”.

JsonStorage encodes the contents of the tuple in JSON format and writes the resulting string into the value field of TextOutputFormat. The key field is left null.

// JsonStorage.java
public void putNext(Tuple t) throws IOException {
    // Build a ByteArrayOutputStream to write the JSON into
    ByteArrayOutputStream baos = new ByteArrayOutputStream(BUF_SIZE);
    // Build the generator
    JsonGenerator json =
        jsonFactory.createJsonGenerator(baos, JsonEncoding.UTF8);

    // Write the beginning of the top level tuple object
    json.writeStartObject();
    for (int i = 0; i < fields.length; i++) {
        writeField(json, fields[i], t.get(i));
    }
    json.writeEndObject();
    json.close();

    // Hand a null key and our string to Hadoop
    try {   
        writer.write(null, new Text(baos.toByteArray()));
    } catch (InterruptedException ie) {
        throw new IOException(ie);
    }
}

private void writeField(JsonGenerator json,
                        ResourceFieldSchema field,
                        Object d) throws IOException {

    // If the field is missing or the value is null, write a null
    if (d == null) {
        json.writeNullField(field.getName());
        return;
    }

    // Based on the field's type, write it out
    switch (field.getType()) {
    case DataType.INTEGER:
        json.writeNumberField(field.getName(), (Integer)d);
        return;

    case DataType.LONG:
        json.writeNumberField(field.getName(), (Long)d);
        return;

    case DataType.FLOAT:
        json.writeNumberField(field.getName(), (Float)d);
        return;

    case DataType.DOUBLE:
        json.writeNumberField(field.getName(), (Double)d);
        return;

    case DataType.BYTEARRAY:
        json.writeBinaryField(field.getName(), ((DataByteArray)d).get());
        return;

    case DataType.CHARARRAY:
        json.writeStringField(field.getName(), (String)d);
        return;

    case DataType.MAP:
        json.writeFieldName(field.getName());
        json.writeStartObject();
        for (Map.Entry<String, Object> e : ((Map<String, Object>)d).entrySet()) {
            json.writeStringField(e.getKey(), e.getValue().toString());
        }
        json.writeEndObject();
        return;

    case DataType.TUPLE:
        json.writeFieldName(field.getName());
        json.writeStartObject();

        ResourceSchema s = field.getSchema();
        if (s == null) {
            throw new IOException("Schemas must be fully specified to use "
                + "this storage function.  No schema found for field " +
                field.getName());
        }
        ResourceFieldSchema[] fs = s.getFields();

        for (int j = 0; j < fs.length; j++) {
            writeField(json, fs[j], ((Tuple)d).get(j));
        }
        json.writeEndObject();
        return;

    case DataType.BAG:
        json.writeFieldName(field.getName());
        json.writeStartArray();
        s = field.getSchema();
        if (s == null) {
            throw new IOException("Schemas must be fully specified to use "
                + "this storage function.  No schema found for field " +
                field.getName());
        }
        fs = s.getFields();
        if (fs.length != 1 || fs[0].getType() != DataType.TUPLE) {
            throw new IOException("Found a bag without a tuple "
                + "inside!");
        }
        // Drill down the next level to the tuple's schema.
        s = fs[0].getSchema();
        if (s == null) {
            throw new IOException("Schemas must be fully specified to use "
                + "this storage function.  No schema found for field " +
                field.getName());
        }
        fs = s.getFields();
        for (Tuple t : (DataBag)d) {
            json.writeStartObject();
            for (int j = 0; j < fs.length; j++) {
                writeField(json, fs[j], t.get(j));
            }
            json.writeEndObject();
        }
        json.writeEndArray();
        return;
    }
}