You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by Apache Wiki <wi...@apache.org> on 2009/12/17 00:23:49 UTC
[Pig Wiki] Update of "LoadStoreRedesignProposal" by RichardDing
Dear Wiki user,
You have subscribed to a wiki page or wiki category on "Pig Wiki" for change notification.
The "LoadStoreRedesignProposal" page has been changed by RichardDing.
http://wiki.apache.org/pig/LoadStoreRedesignProposal?action=diff&rev1=39&rev2=40
--------------------------------------------------
= Proposed Redesign For Load, Store, and Slicer in Pig =
-
== Goals ==
The current design of !LoadFunc, !StoreFunc, and the Slicer interfaces in Pig are not adequate. This proposed redesign has the following goals:
@@ -12, +11 @@
1. These new storage formats also plan to support pushing of, at least, projection and selection into the storage layer. Pig needs to be able to query loaders to determine what if any pushdown capabilities they support and then make use of those capabilities.
1. There already exists one metadata system in Hadoop (Hive's metastore) and there is a proposal to add another (Owl). Pig needs to be able to query these metadata systems for information about data to be read. It also needs to be able to record information to these metadata systems when writing data. The load and store functions are a reasonable place to do these operations since that is the point at which Pig is reading and writing data. This will also allow Pig to read and write data from and to multiple metadata stores in single Pig Latin scripts if that is desired.
+ A requirement for the implementation that does not fit into the goals above is that while the existing Pig implementation is tightly tied to Hadoop (and is becoming more tightly tied all the time), we do not want to tie Pig Latin tightly to Hadoop. Therefore while we plan to allow users to easily interact with Hadoop !InputFormats and !OutputFormats, these should not be exposed as such to Pig Latin. Pig Latin must still view these as load and store functions; it will only be the underlying implementation that will realize that they are Hadoop classes and handle them appropriately.
- A requirement for the implementation that does not fit into the goals above is that while the existing Pig implementation is tightly tied to
- Hadoop (and is becoming more tightly tied all the time), we do not want to tie Pig Latin tightly to Hadoop. Therefore while we plan to allow
- users to easily interact with Hadoop !InputFormats and !OutputFormats, these should not be exposed as such to Pig Latin. Pig Latin must still
- view these as load and store functions; it will only be the underlying implementation that will realize that they are Hadoop classes and handle
- them appropriately.
== Interfaces ==
+ With these proposed changes, load and store functions in Pig are becoming very weighty objects. The current !LoadFunc interface already provides mechanisms for reading the data, getting some schema information, casting data, and some place holders for pushing down projections into the loader. This proposal will add more file level metadata, selection push down, plus interaction with !InputFormats. It will also add !OutputFormats to store functions. If we create two monster interfaces that attempt to provide everything, the burden of creating a new load or store function in Pig will become overwhelming. Instead, this proposal envisions splitting the interface into a number of interfaces, each with a clear responsibility. Load and store functions will then only be required to implement the interfaces for functionality they offer.
- With these proposed changes, load and store functions in Pig are becoming very weighty objects. The current !LoadFunc interface already
- provides mechanisms for reading the data, getting some schema information, casting data, and some place holders for pushing down projections into
- the loader. This proposal will add more file level metadata, selection push down, plus interaction with !InputFormats. It will
- also add !OutputFormats to store functions. If we create two monster interfaces that attempt to provide everything, the burden of creating a
- new load or store function in Pig will become overwhelming. Instead, this proposal envisions splitting the interface into a number of
- interfaces, each with a clear responsibility. Load and store functions will then only be required to implement the interfaces for functionality they offer.
For load functions:
+
* '''!LoadFunc''' will be pared down to just contain functions directly associated with reading data, such as getNext.
* A new '''!LoadCaster''' interface will be added. This interface will contain all of the bytesToX methods currently in !LoadFunc. !LoadFunc will add a `getCaster` routine, that will return an object that can provide casts. The existing UTF8!StorageConverter class will change to implement this interface. Load functions will then be free to use this class as their caster, or provide their own. For existing load functions that provide all of the bytesToX methods, they can implement the !LoadCaster interface and return themselves from the getCaster routine. If a loader does not provide a !LoadCaster, casts from byte array to other Pig types will not be supported for data loaded via that loader.
* A new '''!LoadMetadata''' interface will be added. Calls that find metadata about the data being loaded, such as determineSchema, will be placed in this interface. If a loader does not implement this interface, then Pig will assume that no metadata is obtainable for this data.
* A new '''!LoadPushDown''' interface will be added. Calls that determine what can be pushed down and pushing that functionality down into the loader will be placed in this interface. If a loader does not implement this interface, then Pig will assume that the loader is not capable of pushing down any functionality.
For store functions:
+
* A new method `getOutputFormat` will be added to !StoreFunc to allow a storage function to return its output format.
* A new interface '''!StoreMetadata''' will be added to provide a way for storage functions to record metadata. If a given storage function does not implement this interface Pig will assume that it is unable to record metadata.
=== Details ===
-
'''!LoadFunc'''
{{{
-
/**
* This interface is used to implement functions to parse records
* from a dataset.
@@ -50, +40 @@
/**
* This method is called by the Pig runtime in the front end to convert the
* input location to an absolute path if the location is relative. The
- * loadFunc implementation is free to choose how it converts a relative
+ * loadFunc implementation is free to choose how it converts a relative
* location to an absolute location since this may depend on what the location
* string represent (hdfs path or some other data source)
- *
+ *
* @param location location as provided in the "load" statement of the script
* @param curDir the current working direction based on any "cd" statements
* in the script before the "load" statement. If there are no "cd" statements
- * in the script, this would be the home directory -
+ * in the script, this would be the home directory -
* <pre>/user/<username> </pre>
* @return the absolute location based on the arguments passed
* @throws IOException if the conversion is not possible
@@ -66, +56 @@
/**
- * Communicate to the loader the location of the object(s) being loaded.
+ * Communicate to the loader the location of the object(s) being loaded.
- * The location string passed to the LoadFunc here is the return value of
+ * The location string passed to the LoadFunc here is the return value of
- * {@link LoadFunc#relativeToAbsolutePath(String, String)}
+ * {@link LoadFunc#relativeToAbsolutePath(String, String)}
- *
+ *
* This method will be called in the backend multiple times. Implementations
* should bear in mind that this method is called multiple times and should
* ensure there are no inconsistent side effects due to the multiple calls.
- *
+ *
- * @param location Location as returned by
+ * @param location Location as returned by
* {@link LoadFunc#relativeToAbsolutePath(String, String)}.
* @param job the {@link Job} object
* @throws IOException if the location is not valid.
*/
void setLocation(String location, Job job) throws IOException;
-
+
/**
* This will be called during planning on the front end. This is the
- * instance of InputFormat (rather than the class name) because the
+ * instance of InputFormat (rather than the class name) because the
- * load function may need to instantiate the InputFormat in order
+ * load function may need to instantiate the InputFormat in order
* to control how it is constructed.
* @return the InputFormat associated with this loader.
- * @throws IOException if there is an exception during InputFormat
+ * @throws IOException if there is an exception during InputFormat
* construction
*/
InputFormat getInputFormat() throws IOException;
/**
- * This will be called on the front end during planning and not on the back
+ * This will be called on the front end during planning and not on the back
* end during execution.
- * @return the {@link LoadCaster} associated with this loader. Returning null
+ * @return the {@link LoadCaster} associated with this loader. Returning null
- * indicates that casts from byte array are not supported for this loader.
+ * indicates that casts from byte array are not supported for this loader.
* construction
- * @throws IOException if there is an exception during LoadCaster
+ * @throws IOException if there is an exception during LoadCaster
*/
LoadCaster getLoadCaster() throws IOException;
@@ -122, +112 @@
Tuple getNext() throws IOException;
}
-
}}}
+ The '''!LoadCaster''' interface will include bytesToInt, bytesToLong, etc. functions currently in !LoadFunc. UTF8!StorageConverter will implement this interface.
- The '''!LoadCaster''' interface will include bytesToInt, bytesToLong, etc. functions
- currently in !LoadFunc. UTF8!StorageConverter will implement this interface.
-
- '''Open Question''': Should the methods to convert to a Bag, Tuple and Map take a Schema (ResourceSchema?) argument?
+ '''Open Question''': Should the methods to convert to a Bag, Tuple and Map take a Schema (ResourceSchema?) argument?
-
'''!LoadMetadata'''
- {{{
+ {{{
/**
* 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
@@ -142, +128 @@
public interface LoadMetadata {
/**
- * Get a schema for the data to be loaded.
+ * Get a schema for the data to be loaded.
- * @param location Location as returned by
+ * @param location Location as returned by
* {@link LoadFunc#relativeToAbsolutePath(String, String)}
- * @param conf The {@link Configuration} object
+ * @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
+ 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 as returned by
+ * @param location Location as returned by
* {@link LoadFunc#relativeToAbsolutePath(String, String)}
* @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)
+ ResourceStatistics getStatistics(String location, Configuration conf)
throws IOException;
/**
* Find what columns are partition keys for this input.
- * @param location Location as returned by
+ * @param location Location as returned by
* {@link LoadFunc#relativeToAbsolutePath(String, String)}
* @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)
+ String[] getPartitionKeys(String location, Configuration conf)
throws IOException;
/**
@@ -190, +176 @@
void setParitionFilter(OperatorPlan plan) throws IOException;
}
-
}}}
-
- '''!ResourceSchema''' will be a top level object (`org.apache.pig.ResourceSchema`) used to communicate information about data to be loaded or that is being
+ '''!ResourceSchema''' will be a top level object (`org.apache.pig.ResourceSchema`) used to communicate information about data to be loaded or that is being stored. It is not the same as the existing `org.apache.pig.impl.logicalLayer.schema.Schema`.
- stored. It is not the same as the existing `org.apache.pig.impl.logicalLayer.schema.Schema`.
{{{
public class ResourceSchema {
@@ -212, +195 @@
enum Order { ASCENDING, DESCENDING }
public int[] sortKeys; // each entry is an offset into the fields array.
- public Order[] sortKeyOrders;
+ public Order[] sortKeyOrders;
}
}}}
-
- Feedback from Pradeep: We must fix the two level access issues with schema of bags in current schema before we make these changes, otherwise that
+ Feedback from Pradeep: We must fix the two level access issues with schema of bags in current schema before we make these changes, otherwise that same contagion will afflict us here.
- same contagion will afflict us here.
'''!ResourceStatistics'''
+
{{{
public class ResourceStatistics {
@@ -247, +229 @@
// Probably more in here
}
}}}
+ At this point, !ResourceStatistics is poorly understood. In initial versions we may choose not to implement it. In additions to questions on what should be in the statistics, there are questions on how statistics should be communicated in relation to partitions. For example, when loading from a table that is stored in owl or hive, one or more partitions may be being loaded. Assuming statistics are kept on the partition level, how are these statistics then communicated to Pig? Is it the loader's job to combine the statistics for all of the partitions being read? Or does it return an array of !ResourceStatistics? But if so, what does Pig do with it since it does not know which tuples belong to which partitions (and doesn't want to know). Even worse on store, any statistics Pig has to report is across all data being stored. But the storage function underneath may choose to partition the data. How does it then separate those statistics for the different partitions? In these cases should store functions be in charge of calculating statistics? Perhaps some statistics that can be easily distributed across partitions (such as distribution types) should be calculated and sent down by Pig and some left to the store functions. Perhaps statistics on store should instead be driven by call backs that Pig could define and implement and store functions should call after data has been partitioned if they want to.
-
- At this point, !ResourceStatistics is poorly understood. In initial versions we may choose not to implement it. In additions to questions
- on what should be in the statistics, there are questions on how statistics should be communicated in relation to partitions. For example,
- when loading from a table that is stored in owl or hive, one or more partitions may be being loaded. Assuming statistics are kept on the
- partition level, how are these statistics then communicated to Pig? Is it the loader's job to combine the statistics for all of the
- partitions being read? Or does it return an array of !ResourceStatistics? But if so, what does Pig do with it since it does not know which
- tuples belong to which partitions (and doesn't want to know). Even worse on store, any statistics Pig has to report is across all data
- being stored. But the storage function underneath may choose to partition the data. How does it then separate those statistics for the
- different partitions? In these cases should store functions be in charge of calculating statistics? Perhaps some statistics that can be
- easily distributed across partitions (such as distribution types) should be calculated and sent down by Pig and some left to the store
- functions. Perhaps statistics on store should instead be driven by call backs that Pig could define and implement and store functions should
- call after data has been partitioned if they want to.
-
'''!LoadPushdown'''
- {{{
+ {{{
/**
* This interface defines how to communicate to Pig what functionality can
* be pushed into the loader. If a given loader does not implement this interface
@@ -272, +242 @@
interface LoadPushDown {
/**
- * Set of possible operations that Pig can push down to a loader.
+ * Set of possible operations that Pig can push down to a loader.
*/
enum OperatorSet {PROJECTION, SELECTION};
/**
- * Determine the operators that can be pushed to the loader.
+ * Determine the operators that can be pushed to the loader.
* Note that by indicating a loader can accept a certain operator
* (such as selection) the loader is not promising that it can handle
- * all selections. When it is passed the actual operators to
+ * all selections. When it is passed the actual operators to
* push down it will still have a chance to reject them.
* @return list of all features that the loader can support
*/
@@ -295, +265 @@
*/
boolean pushOperators(OperatorPlan plan);
}
-
}}}
+ An open question for !LoadPushdown is how do we communicate what needs to be pushed down? In the above, !OperatorPlan is envisioned as a simple syntax tree that would be sufficient to communicate operators and their expressions to the storage functions. Initially we considered using the !LogicalPlan as is. But our !LogicalPlan is a mess right now. It's also tightly tied to our implementation, so exposing it as an interface is unwise. We could use a SQL like string, but then loaders have to implement a parser.
-
- An open question for !LoadPushdown is how do we communicate what needs to be pushed down? In the above, !OperatorPlan is envisioned as a simple
- syntax tree that would be sufficient to communicate operators and their expressions to the storage functions. Initially we considered using the
- !LogicalPlan as is. But our !LogicalPlan is a mess right now. It's also tightly tied to our implementation, so exposing it as an interface is
- unwise. We could use a SQL like string, but then loaders have to implement a parser.
'''!StoreFunc'''
{{{
-
/**
* This interface is used to implement functions to write records
* from a dataset.
@@ -317, +281 @@
/**
* This method is called by the Pig runtime in the front end to convert the
* output location to an absolute path if the location is relative. The
- * StoreFunc implementation is free to choose how it converts a relative
+ * StoreFunc implementation is free to choose how it converts a relative
* location to an absolute location since this may depend on what the location
* string represent (hdfs path or some other data source)
- *
+ *
* @param location location as provided in the "store" statement of the script
* @param curDir the current working direction based on any "cd" statements
* in the script before the "store" statement. If there are no "cd" statements
- * in the script, this would be the home directory -
+ * in the script, this would be the home directory -
* <pre>/user/<username> </pre>
* @return the absolute location based on the arguments passed
* @throws IOException if the conversion is not possible
@@ -334, +298 @@
/**
* Return the OutputFormat associated with StoreFunc. This will be called
* on the front end during planning and not on the backend during
- * execution.
+ * execution.
* @return the {@link OutputFormat} associated with StoreFunc
- * @throws IOException if an exception occurs while constructing the
+ * @throws IOException if an exception occurs while constructing the
* OutputFormat
*
*/
OutputFormat getOutputFormat() throws IOException;
/**
- * Communicate to the store function the location used in Pig Latin to refer
+ * Communicate to the store function the location used in Pig Latin to refer
* to the object(s) being stored. That is, if the PL script is
* <b>store A into 'bla'</b>
* then 'bla' is the location. This location should be either a file name
- * or a URI. If it does not have a URI scheme Pig will assume it is a
+ * or a URI. If it does not have a URI scheme Pig will assume it is a
- * filename. This will be
+ * filename. This will be
* called during planning on the front end, not during execution on
* the backend.
* @param location Location indicated in store statement.
@@ -356, +320 @@
* @throws IOException if the location is not valid.
*/
void setStoreLocation(String location, Job job) throws IOException;
-
+
/**
* Set the schema for data to be stored. This will be called on the
* front end during planning. A Store function should implement this function to
* check that a given schema is acceptable to it. For example, it
* can check that the correct partition keys are included;
* a storage function to be written directly to an OutputFormat can
- * make sure the schema will translate in a well defined way.
+ * make sure the schema will translate in a well defined way.
* @param s to be checked
* @throws IOException if this schema is not acceptable. It should include
* a detailed error message indicating what is wrong with the schema.
@@ -381, +345 @@
/**
* Write a tuple the output stream to which this instance was
* previously bound.
- *
+ *
* @param t the tuple to store.
* @throws IOException if an exception occurs during the write
*/
void putNext(Tuple t) throws IOException;
}
-
}}}
-
'''!StoreMetadata'''
{{{
@@ -409, +371 @@
/**
* Set schema of the data being written
- * @throws IOException
+ * @throws IOException
*/
void setSchema(ResourceSchema schema, String location, Configuration conf) throws IOException;
}
-
}}}
-
Given the uncertainly noted above under !ResourceStatistics on how statistics should be stored, it is not clear that this interface makes sense.
== LoadFunc and InputFormat Interaction ==
+ The Slicer and Slice interfaces duplicate much of the !InputFormat and !InputSplit interfaces provided by Hadoop. The current !SliceWrapper helper function interacts with these slices to encode information about input sources into each !InputSplit so that Pig can correctly manage which processing pipeline a tuple from a given input is assigned to. Currently load functions also duplicate much of the functionality of Hadoop's !RecordReader interface. Currently load functions are provided a data stream and asked to parse both records and fields within records. !RecordReader already provides the functionality to parse out records, though it always returns two values, key and value, not a tuple of any number of values. An integral part of this proposal is to change the !LoadFunc to call !RecordReader.getCurrentKey and getCurrentValue and then parse out fields in the tuple from that result.
+ Since Pig still needs to add information to !InputSplits, user provided !InputFormats and !InputSplits cannot be used directly. Instead, the proposal is to change !PigInputFormat to represent the job's !InputFormat to !Hadoop and internally to handle the complexity of multiple inputs and hence multiple !InputFormats. !PigInputFormat will return !PigSplits each of which contain an !InputSplit. In addition, !PigSplit will contain the necessary information to allow Pig to correctly address tuples to the correct data processing pipeline.
- The Slicer and Slice interfaces duplicate much of the !InputFormat and !InputSplit interfaces provided by Hadoop. The current !SliceWrapper
- helper function interacts with these slices to encode information about input sources into each !InputSplit so that Pig can correctly manage
- which processing pipeline a tuple from a given input is assigned to. Currently load functions also duplicate much of the functionality of
- Hadoop's !RecordReader interface. Currently load functions are provided a data stream and asked to parse both records and fields within records.
- !RecordReader already provides the functionality to parse out records, though it always returns two values, key and value, not a tuple of any
- number of values. An integral part of
- this proposal is to change the !LoadFunc to call !RecordReader.getCurrentKey and getCurrentValue and then parse out fields in the tuple from that
- result.
+ In order to support arbitrary Hadoop !InputFormats, Pig can provide a load function, !InputFormatLoader, that will take an !InputFormat as a constructor argument. Only !InputFormats which have zero argument constructors can be supported since Pig will try to instantiate the supplied !InputFormat using reflection. When asked by Pig which !InputFormat to use, it will return the one indicated by the user. Its call to getNext will then take the key and value provided by the associated !RecordReader and construct a two field tuple. These types will be converted to Pig types as follows:
+ ||Writable ||Pig type ||Comment ||
+ ||Text ||chararray || ||
+ ||!IntWritable ||integer || ||
+ ||V!IntWritable ||integer || ||
+ ||!LongWritable ||long || ||
+ ||V!LongWritable ||long || ||
+ ||!FloatWritable ||float || ||
+ ||!DoubleWritable ||double || ||
- Since Pig still needs to add information to !InputSplits, user provided !InputFormats and !InputSplits cannot be used directly. Instead, the
- proposal is to change !PigInputFormat to represent the job's !InputFormat to !Hadoop and internally to handle the complexity of multiple inputs and hence multiple !InputFormats. !PigInputFormat will return !PigSplits each of which contain an
- !InputSplit. In addition, !PigSplit will contain the necessary information to allow Pig to correctly address tuples to the correct data
- processing pipeline.
-
- In order to support arbitrary Hadoop !InputFormats, Pig can provide a load function, !InputFormatLoader, that will take an
- !InputFormat as a constructor argument. Only !InputFormats which have zero argument constructors can be supported since Pig will try to instantiate the supplied !InputFormat using reflection. When asked by Pig which !InputFormat to use, it will return the one indicated by the user. Its call to
- getNext will then take the key and value provided by the associated !RecordReader and construct a two field tuple. These types will be converted
- to Pig types as follows:
-
- || Writable || Pig type || Comment ||
- || Text || chararray || ||
- || !IntWritable || integer || ||
- || V!IntWritable || integer || ||
- || !LongWritable || long || ||
- || V!LongWritable || long || ||
- || !FloatWritable || float || ||
- || !DoubleWritable || double || ||
- || !BooleanWritable || int || In the future if Pig exposes boolean as a first class type, this would change to boolean ||
+ ||!BooleanWritable ||int ||In the future if Pig exposes boolean as a first class type, this would change to boolean ||
- || !ByteWritable || int || ||
- || !NullWritable || null || ||
+ ||!ByteWritable ||int || ||
+ ||!NullWritable ||null || ||
- || All others || byte array || How do we construct a byte array from arbitrary types? ||
+ ||All others ||byte array ||How do we construct a byte array from arbitrary types? ||
- Since the format of any other types are unknown to Pig and cannot be generalized, it does not make sense to provide casts from byte array to pig
- types via a !LoadCaster. If users wish to use an !InputFormat that uses types beyond these and cast them to Pig types, they can extend the
- !InputFormatLoader and return a custom written !LoadCaster that will handle casting their data types.
+
+
+
+ Since the format of any other types are unknown to Pig and cannot be generalized, it does not make sense to provide casts from byte array to pig types via a !LoadCaster. If users wish to use an !InputFormat that uses types beyond these and cast them to Pig types, they can extend the !InputFormatLoader and return a custom written !LoadCaster that will handle casting their data types.
=== Details ===
+ Currently, in addition to information regularly passed with an !InputSplit, Pig also passes along additional information via the !SliceWrapper (Pig's implementation of !InputSplit). This includes an index (it's not clear what this is for, and there are no calls to !SliceWrapper.getIndex() in the code; boy comments in code sure are nice), and exec type to indicate MR or local mode, and the list of target operators (the pipeline) for this input. These items can continue to be carried along as is in the proposed !PigInputSplit.
- Currently, in addition to information regularly passed with an !InputSplit, Pig also passes along additional information via the !SliceWrapper
- (Pig's implementation of !InputSplit). This includes an index (it's not clear what this is for, and there are no calls to
- !SliceWrapper.getIndex() in the code; boy comments in code sure are nice), and exec type to indicate MR or local mode, and the list of target
- operators (the pipeline) for this input. These items can continue to be carried along as is in the proposed !PigInputSplit.
Pig's Slice interface currently supports the following calls:
+
* getLocations - will be replaced by !InputSplit.getLocations
* init - replaced by !RecordReader.initialize
* getStart - see below
@@ -473, +417 @@
* getProgress - replaced by !RecordReader.getProgress
* next - replaced by !RecordReader.getCurrentKey and getCurrentValue
+ Positioning information in an !InputSplit presents a problem. Hadoop 0.18 has a getPos call in the !InputSplit, but it has been removed in 0.20. The reason is that input from files can generally be assigned a position, though it may not always be accurate, as in the bzip case. But for some input formats position may not have meaning. Even if Pig does not switch to using !InputFormats it will have to deal with this issue, just as MR has.
- Positioning information in an !InputSplit presents a problem. Hadoop 0.18 has a getPos call in the !InputSplit, but it has been removed in 0.20.
- The reason is that input from files can generally be assigned a position, though it may not always be
- accurate, as in the bzip case. But for some input formats position may not have meaning. Even if Pig does not switch to using !InputFormats it will
- have to deal with this issue, just as MR has.
+ These changes will affect the !SamplableLoader interface. Currently it uses skip and getPos to move the underlying stream so that it can pick up a sample of tuples out of a block. Since it would sit atop !InputFormat it would no longer have access to the underlying stream. It would be changed instead to skip a number of tuples.
- These changes will affect the !SamplableLoader interface. Currently it uses skip and getPos to move the underlying stream so that it can pick
- up a sample of tuples out of a block. Since it would sit atop !InputFormat it would no longer have access to the underlying stream. It would be
- changed instead to skip a number of tuples.
- However, in some places Pig needs this position information. In particular, when building an index for a merge join, Pig needs a way to mark a
- location in an input while building the index and then return to that position during the join. In this new proposal, the merge join index will contain filename and split index (index of the split in the List returned by InputFormat.getSplits()). The merge join code at run time will then seek to the right split in the file and process from that split on. For this to work the assumption is if we start from a split and read from there on to the last splitwe get sorted data - i.e. the splits in getSplits() is preserve ordering. Since this in general cannot be guaranteed, the proposal to handle this is to sample the first and last keys in each split and record both values in the index entry for that split. The index is then sorted based on both the first and last key. While seeking into the right file based on the join key during merge join processing, the implementation will then read the relevant splits in the right file as indicated in the index (reading the splits from the matching index entry to the last index entry).
+ However, in some places Pig needs this position information. In particular, when building an index for a merge join, Pig needs a way to mark a location in an input while building the index and then return to that position during the join. In this new proposal, the merge join index will contain filename and split index (index of the split in the List returned by InputFormat.getSplits()). The merge join code at run time will then seek to the right split in the file and process from that split on. For this to work the assumption is if we start from a split and read from there on to the last splitwe get sorted data - i.e. the splits in getSplits() is preserve ordering. Since this in general cannot be guaranteed, the proposal to handle this is to sample the first and last keys in each split and record both values in the index entry for that split. The index is then sorted based on both the first and last key. While seeking into the right file based on the join key during merge join processing, the implementation will then read the relevant splits in the right file as indicated in the index (reading the splits from the matching index entry to the last index entry).
+ These changes will also affect loaders that need to read a record in order to determine their schema, such as !BinStorage or a JSON loader. These loaders will need to know how to read at least the first record on their own, without the benefit of the underlying !InputFormat, since they will need to call this on the front end where an !InputFormat has not yet been used.
- These changes will also affect loaders that need to read a record in order to determine their schema, such as !BinStorage or a JSON loader.
- These loaders will need to know how to read at least the first record on their own, without the benefit of the underlying !InputFormat, since
- they will need to call this on the front end where an !InputFormat has not yet been used.
+ In addition to opening files as part of Map-Reduce, Pig loaders also open files on the side in MR jobs. The new load interface needs to be able to open these side files as well. For this we would need to create a new instance of the appropriate !InputFormat, calls getSplits, and then iterate over the split and for each split create a RecordReader and process the data returned by the RecordReader and then move to the next split.
- In addition to opening files as part of Map-Reduce, Pig loaders also open files on the side in MR jobs. The new load interface needs to be able to
- open these side files as well. For this we would need to create a new instance of the appropriate !InputFormat, calls getSplits, and
- then iterate over the split and for each split create a RecordReader and process the data returned by the RecordReader and then move to the next split.
'''Performance concerns.'''
+ With this proposal we are strongly encouraging users to double parse their data (since they will parse out the records and then parse those into fields). This could carry a performance penalty. Also, we are removing the ability to seek when sampling, which could also contain a performance penalty. A few thoughts on this:
+
- With this proposal we are strongly encouraging users to double parse their data (since they will parse out the records and then parse those into
- fields). This could carry a performance penalty. Also, we are removing the ability to seek when sampling, which could also contain a
- performance penalty. A few thoughts on this:
1. We are already in the process of moving !PigStorage to use !LineRecordReader underneath, and we are getting a 30% speed up for doing it. This is only one case, but at least in this case double parsing is not hurting us.
1. If users that are writing both an !InputFormat and a load function are concerned about double parsing, there is nothing stopping them from having their !InputFormat track where field boundaries are and pass that information along to their loader so that they can parse in a single pass.
1. For sampling, it is not clear how sever the penalty is for removing seek, since Hadoop is reading much of the information off disk into the buffers anyway, and sequential read on disk is not that much worse than seek.
- All this said, before committing to this we should do some prototyping and performance testing to convince ourselves that these changes will be
+ All this said, before committing to this we should do some prototyping and performance testing to convince ourselves that these changes will be performance neutral.
- performance neutral.
== StoreFunc and OutputFormat Interaction ==
In the same way that !LoadFunc currently duplicates some functionality of !InputFormat, !StoreFunc duplicates some functionality of !OutputFormat. !StoreFunc will be changed to deal primarily with converting a tuple to a key value pair that can be stored by Hadoop.
+ To support arbitrary !OutputFormats, a new storage function !OutputFormatStorage could be written that will take an !OutputFormat as a constructor argument. Only !OutputFormats which have zero argument constructors can be supported since Pig will try to instantiate the supplied !OutputFormat using reflection. Tuples to be stored by this storage function must have either one or two fields. If they have two fields, the first of will be taken to be the key, and the second the value. If they have one, the key will be set to null and the value will be taken from the single field. Data type conversion on this data will be done in the same way as noted above for !InputFormatLoader.
- To support arbitrary !OutputFormats, a new storage function !OutputFormatStorage could be written that will take an !OutputFormat as a constructor
- argument. Only !OutputFormats which have zero argument constructors can be supported since Pig will try to instantiate the supplied !OutputFormat using reflection. Tuples to be stored by this storage function must have either one or two fields. If they have two fields, the first of will be taken
- to be the key, and the second the value. If they have one, the key will be set to null and the value will be taken from the single field. Data
- type conversion on this data will be done in the same way as noted above for !InputFormatLoader.
Open Questions:
+
1. Does all this force us to switch to Hadoop for local mode as well? We aren't opposed to using Hadoop for local mode it just needs to get reasonable fast. Can we use !InputFormat ''et. al.'' on local files without using the whole HDFS structure? '''Answer''' According to Hadoop documentation !TextInputFormat works on local files as well as hdfs files. We may need to catch that we are in local mode and change the filename to `file://` OR change to using Hadoop's local mode
1. How will we worked with compressed files? !FileInputFormat already works with bzip and gzip compressed files, producing reasonable splits. !PigStorage will be reworked to depend on !FileInputFormat (or a descendant thereof, see next item) and should therefore be able to use this functionality. Currently Pig supports gz/bzip for arbitrary loadfunc/storefunc combinations. With this proposal, gz/bzip format will only be supported for load/store using PigStorage.
-
== Implementation details and status ==
-
=== Current status ===
- https://issues.apache.org/jira/browse/PIG-966 is the main JIRA to track progress. A branch -'load-store-redesign' (http://svn.apache.org/repos/asf/hadoop/pig/branches/load-store-redesign) has been created to undertake work on this proposal.
+ https://issues.apache.org/jira/browse/PIG-966 is the main JIRA to track progress. A branch -'load-store-redesign' (http://svn.apache.org/repos/asf/hadoop/pig/branches/load-store-redesign) has been created to undertake work on this proposal.
Status on Nov 2. 2009: This branch has simple load-store working for !PigStorage and !BinStorage. Joins on multiple inputs and multi store queries with multi query optimization also work. Some of the recent changes in the proposal above (the changes noted under Nov 2. 2009 in the Changes below) have not been incorporated. A list (may not be comprehensive) of remaining tasks is listed in a subsection below.
=== Notes on implementation details ===
- This section is to document changes made at a high level to give an overall connected picture which code comments may not provide.
+ This section is to document changes made at a high level to give an overall connected picture which code comments may not provide.
==== Changes to work with Hadoop InputFormat model ====
Hadoop has the notion of a single InputFormat per job. This is restrictive since Pig processes multiple inputs in the same map reduce job (in the case of Join, Union or Cogroup). This is handled by !PigInputFormat which is the !InputFormat Pig communicates to Hadoop as the Job's !InputFormat. In !PigInputFormat.getSplits(), the implementation processes each input in the following manner:
@@ -543, +471 @@
* Instantiate the !LoadFunc associated with input represented by the PigSplit passed into !PigInputFormat.createRecordReader()
* invoke !LoadFunc.setLocation()
* Call getInputFormat() on the !LoadFunc and then createRecordReader() on the !InputFormat returned. Note that the above setLocation call needs to happen *before* the createRecordReader() call and the createRecordReader() call needs to be given a !TaskAttemptContext built out of the "updated (with location)" Configuration.
- * Wrap the !RecordReader returned above in !PigRecordReader class which is returned to Hadoop as the !RecordReader. !PigRecordReader has Text as key type (which is always sent with a null value to Hadoop since in pig, we really do not extract a key from input records) and a Tuple as a the value type (which is a tuple constructed from the input record).
+ * Wrap the !RecordReader returned above in !PigRecordReader class which is returned to Hadoop as the !RecordReader. !PigRecordReader has Text as key type (which is always sent with a null value to Hadoop since in pig, we really do not extract a key from input records) and a Tuple as a the value type (which is a tuple constructed from the input record).
'''Open Question''': - We are hoping that !LoadFunc actually sets up the input location on the conf in the setLocation() call - and then using that conf in createRecordReader() call - what if it does this in getInputFormat()?
==== Changes to work with Hadoop OutputFormat model ====
- Hadoop has the notion of a single !OutputFormat per job. !PigOutputFormat is the class indicated by Pig as the !OutputFormat for map reduce jobs compiled from pig scripts.
+ Hadoop has the notion of a single !OutputFormat per job. !PigOutputFormat is the class indicated by Pig as the !OutputFormat for map reduce jobs compiled from pig scripts.
In !PigOutputFormat.checkOutputSpecs(), the implementation iterates over !POStore(s) in the map and reduce phases and for each such store does the following:
+
* Instantiate the !StoreFunc associated with the !POStore
* Make a clone of the !JobContext passed in !PigOutputFormat.checkOutputSpecs() call and then invoke !StoreFunc.setStoreLocation() using the clone. The reason a clone is necessary here is because generally in the setStorelocation() method, the !StoreFunc would communicate the location to its underlying !OutputFormat. Typically !OutputFormats store the location into the Configuration for use in the checkOutputSpecs() call. For example, !FileOutputFormat does this through !FileOutputFormat.setOutputPath(Job job, Path location). We don't want updates to the Configuration for different outputs to over-write each other - hence the clone.
* Call getOutputFormat() on the !StoreFunc and then checkOutputSpecs() on the !OutputFormat returned. Note that the above setStoreLocation call needs to happen *before* the checkOutputSpecs() call and the checkOutputSpecs() call needs to be given the "updated (with location)" cloned JobContext.
@@ -558, +487 @@
!PigOutputFormat.getOutputCommitter() returns a !PigOutputCommitter object. The !PigOutputCommitter internally keeps a list of OutputCommitters corresponding to !OutputFormat of !StoreFunc(s) in the POStore(s) in the map and reduce phases. It delegates all calls in the OutputCommitter class invoked by Hadoop to calls on the appropriate underlying committers.
The other method in !OutputFormat is the getRecordWriter() method. In the single store case !PigOutputFormat.getRecordWriter() does the following:
+
* Instantiate the !StoreFunc associated with single !POStore.
* invoke !StoreFunc.setStoreLocation()
* Call getOutputFormat() on the !StoreFunc and then getRecordWriter() on the !OutputFormat returned. Note that the above setStoreLocation call needs to happen *before* the getRecordWriter() call and the getRecordWriter() call needs to be given a !TaskAttemptContext which has the "updated (with location)" Configuration.
@@ -571, +501 @@
==== Changes to skew join sampling (PoissonSampleLoader) ====
See discussion in [[https://issues.apache.org/jira/browse/PIG-1062|PIG-1062]] .
- '''Problem 1''':
- Earlier version of !PoissonSampleLoader stored the size on disk as an extra last column in the sampled tuples it returned in map phase of sampling MR job. This was used in !PartitionSkwewedKeys udf in the reduce stage of sampling job to compute total number of tuples using input-file-size/avg-disk-sz-from-samples . Avg-disk-sz-from-samples is not available with new loader design, because getPosition() is not there.
+ '''Problem 1''': Earlier version of !PoissonSampleLoader stored the size on disk as an extra last column in the sampled tuples it returned in map phase of sampling MR job. This was used in !PartitionSkwewedKeys udf in the reduce stage of sampling job to compute total number of tuples using input-file-size/avg-disk-sz-from-samples . Avg-disk-sz-from-samples is not available with new loader design, because getPosition() is not there.
- '''Solution :'''
- !PoissonSampleLoader returns a special tuple with number of rows in that Map, in addition to the sampled tuples. To create this special tuple, the max row length in input sampled tuples is tracked, and a new tuple with size of max_row_length + 2 is created.
+ '''Solution :''' !PoissonSampleLoader returns a special tuple with number of rows in that Map, in addition to the sampled tuples. To create this special tuple, the max row length in input sampled tuples is tracked, and a new tuple with size of max_row_length + 2 is created. And spl_tuple[max_row_length ] = "marker_string"
- And spl_tuple[max_row_length ] = "marker_string"
+
- spl_tuple[max_row_length + 1] = num_rows
+ . spl_tuple[max_row_length + 1] = num_rows
+
- The size of max_row_length+2 is used because the join key can be an expression, which is evaluated on the columns in tuples returned by the sampler, and the expression might expect specific data types to be present in certain (<= max_row_length) locations of the tuple.
+ The size of max_row_length+2 is used because the join key can be an expression, which is evaluated on the columns in tuples returned by the sampler, and the expression might expect specific data types to be present in certain (<= max_row_length) locations of the tuple. If number of tuples in sample is 0, the special tuple is not sent.
- If number of tuples in sample is 0, the special tuple is not sent.
In !PartitionSkwewedKeys udf in the reduce stage,the udf iterates over the tuples to find these special tuples and calculate the total number of rows.
+ '''Problem 2''': !PoissonSampleLoader samples 17 tuples from every set of tuples that will fit into reducer memory (see PigSkewedJoinSpec) . Let us call this number of tuples that fit into reducer memory - X. Ie we need to sample one tuple every X/17 tuples. Earlier, the number of tuples to be sampled was calculated before the tuples were read, in !PoissonSampleLoader.computeSamples(..) . To get the number of samples to be sampled in a map, the formula used was = number-of-reducer-memories-needed * 17 / number-of-splits <<BR>> Where - number-of-reducer-memories-needed = (total_file_size * disk_to_mem_factor)/available_reducer_heap_size<<BR>> disk_to_mem_factor has default of 2.
-
- '''Problem 2''':
- !PoissonSampleLoader samples 17 tuples from every set of tuples that will fit into reducer memory (see PigSkewedJoinSpec) . Let us call this number of tuples that fit into reducer memory - X. Ie we need to sample one tuple every X/17 tuples.
- Earlier, the number of tuples to be sampled was calculated before the tuples were read, in !PoissonSampleLoader.computeSamples(..) . To get the number of samples to be sampled in a map, the formula used was = number-of-reducer-memories-needed * 17 / number-of-splits <<BR>>
- Where -
- number-of-reducer-memories-needed = (total_file_size * disk_to_mem_factor)/available_reducer_heap_size<<BR>>
- disk_to_mem_factor has default of 2.
Then !PoissonSampleLoader would return sampled tuples by skipping split-size/num_samples bytes at a time.
- With new loader we have to skip some number of tuples instead of bytes. But we don't have an estimate of total number of tuples in the input.<<BR>>
- One way to work around this would be to use size of tuple in memory to estimate size of tuple in disk using above disk_to_mem_factor, then number of tuples to be skipped will be = (split-size/avg_mem_size_of_tuple)/numSamples
+ With new loader we have to skip some number of tuples instead of bytes. But we don't have an estimate of total number of tuples in the input.<<BR>> One way to work around this would be to use size of tuple in memory to estimate size of tuple in disk using above disk_to_mem_factor, then number of tuples to be skipped will be = (split-size/avg_mem_size_of_tuple)/numSamples
But the use of disk_to_mem_factor is very dubious, the real disk_to_mem_factor will vary based on compression-algorithm, data characteristics (sorting etc), and encoding.
+ '''Solution''': The goal is to sample one tuple every X/17 tuples. (X = number of tuples that fit in available reducer memory).<<BR>> To estimate X, we can use available_reducer_heap_size/average-tuple-mem-size.<<BR>> Number of tuples skipped for every sampled tuple = 1/17 * ( available_reducer_heap_size/average-tuple-mem-size)
- '''Solution''':
- The goal is to sample one tuple every X/17 tuples. (X = number of tuples that fit in available reducer memory).<<BR>>
- To estimate X, we can use available_reducer_heap_size/average-tuple-mem-size.<<BR>>
- Number of tuples skipped for every sampled tuple = 1/17 * ( available_reducer_heap_size/average-tuple-mem-size)
The average-tuple-mem-size and number-of-tuples-to-be-skippled-every-sampled-tuple is recalculated after a new tuple is sampled.
==== Changes to order-by sampling (RandomSampler) ====
+ '''Problem''': With new interface, we cannot use the old approach : In each map job of the sampling MR job, divide the split-size by number of samples required to determine the number of bytes to skip after each sample. <<br>> With new interface, we can skip only by number of tuples not by bytes on disk.
- '''Problem''': With new interface, we cannot use the old approach :
- In each map job of the sampling MR job, divide the split-size by number of samples required to determine the number of bytes to skip after each sample. <<br>>
- With new interface, we can skip only by number of tuples not by bytes on disk.
-
-
- '''Proposal''':
- The goal is to sample tuples with equal probability for any tuple getting sampled (assuming number of tuples to be sampled is much smaller than total number of tuples).
+ '''Proposal''': The goal is to sample tuples with equal probability for any tuple getting sampled (assuming number of tuples to be sampled is much smaller than total number of tuples).
+
- If N is the number of samples required. In getNext(),allocate a buffer for N elements, populate it with the first N tuples, and continue scanning the partition. For every ith next() call, generate a random number r s.t. 0<=r<i, and if r<N insert the new tuple into our buffer at position r. This gives a nicely random sample of the tuples in the partition.
+ . If N is the number of samples required. In getNext(),allocate a buffer for N elements, populate it with the first N tuples, and continue scanning the partition. For every ith next() call, generate a random number r s.t. 0<=r<i, and if r<N insert the new tuple into our buffer at position r. This gives a nicely random sample of the tuples in the partition.
=== Changes to Streaming ===
+ Currently Pig streaming uses store and load functions to handle the serialization and deserialization of data between Pig and the user's executable. Based on the changes proposed in this redesign that will no longer be feasible. Hadoop's !InputFormat and !OutputFormat are not appropriate abstractions for communicating with an external executable.
- Currently Pig streaming uses store and load functions to handle the serialization and deserialization of data between Pig and the user's executable. Based on the changes proposed in this
- redesign that will no longer be feasible. Hadoop's !InputFormat and !OutputFormat are not appropriate abstractions for communicating with an external executable.
+ Pig's current support for variable data formats in streaming is more extensive than Hadoop's. Conceptually users could create executables that read data from !BinStorage or other non-text formats. The Pig development team are not aware of any users doing this, nor is it clear why they would choose to. Removing this functionality will make Pig's streaming implementation simpler. It also avoids shoe-horning streaming into the new load store design. (For complete details of Hadoop's streaming see http://hadoop.apache.org/common/docs/r0.20.1/streaming.html .)
- Pig's current support for variable data formats in streaming is more extensive than Hadoop's. Conceptually users could create executables that read data from !BinStorage or other non-text
- formats. The Pig development team are not aware of any users doing this, nor is it clear why they would choose to. Removing this functionality will make Pig's streaming implementation simpler.
- It also avoids shoe-horning streaming into the new load store design. (For complete details of Hadoop's streaming see http://hadoop.apache.org/common/docs/r0.20.1/streaming.html .)
+ I propose that Pig move to a model of using Hadoop's default streaming format, which is to expect new-line separated records, with tab being used as a field separator. Hadoop allows users to redefine the field separator, and so should Pig. This will also match the current default of using !PigStorage as the (de)serializer for streaming. As before Pig should support communicating with the executable via either stdin and stdout or files. This will force a syntax change in Pig Latin. Currently, if a user wants to stream data to an executable with comma separated fields instead of tab separated fields, the syntax is:
- I propose that Pig move to a model of using Hadoop's default streaming format, which is to expect new-line separated records, with tab being used as a field separator. Hadoop allows users to
- redefine the field separator, and so should Pig. This will also match the current default of using !PigStorage as the (de)serializer for streaming. As before Pig should support communicating
- with the executable via either stdin and stdout or files. This will force a syntax change in Pig Latin. Currently, if a user wants to stream data to an executable with comma separated fields
- instead of tab separated fields, the syntax is:
{{{
- define CMD `perl PigStreaming.pl - foo nameMap` input(stdin using PigStorage(',') output(stdout using PigStorage(',');
+ define CMD `perl PigStreaming.pl - foo nameMap` input(stdin using PigStorage(',')) output(stdout using PigStorage(','));
A = load 'file';
B = stream B through CMD;
}}}
+ The syntax should change the reference to a store and load functions to PigToStream and StreamToPig functions as defined below. Thus the above would become:
- The syntax should change to remove the reference to a store and load functions, as they are no longer meaningful. Thus the above would become:
-
{{{
- define CMD `perl PigStreaming.pl - foo nameMap` input(stdin using ',') output(stdout using ',');
+ define CMD `perl PigStreaming.pl - foo nameMap` input(stdin using PigStreaming(',')) output(stdout using PigStreaming(','));
A = load 'file';
B = stream B through CMD;
}}}
+ From an implementation viewpoint, the functionality required to write to and read from the streaming binary will be equivalent to the tuple parsing and serialization of !PigStorage.getNext() and !PigStorage.putNext(). While it will not be possible to use PigStorage directly every effort should be made to share this code (most likely by putting the actual code in static utility methods that can be called by each class) to avoid double code maintenance costs.
-
- From an implementation viewpoint, the functionality required to write to and read from the streaming binary will be equivalent to the tuple parsing and serialization of !PigStorage.getNext() and
- !PigStorage.putNext(). While it will not be possible to use PigStorage directly every effort should be made to share this code (most likely by putting the actual code in static
- utility methods that can be called by each class) to avoid double code maintenance costs.
It has been suggested that we should switch to the typed bytes protocol that is available in Hadoop and Hive (see
- https://issues.apache.org/jira/browse/PIG-966?focusedCommentId=12781695&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12781695 ). While we cannot switch the default, we can make this streaming
+
+ . [[https://issues.apache.org/jira/browse/PIG-966?focusedCommentId=12781695&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#action_12781695|https://issues.apache.org/jira/browse/PIG-966?focusedCommentId=12781695&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12781695]] ). While we cannot switch the default, we can make this streaming
+
connection an interface so that users can easily extend it in the future. The interface should be quite simple:
{{{
@@ -659, +565 @@
* executable.
*/
public byte[] serialize(Tuple t) throws IOException;
-
- /**
- * Set the record delimiter to use when communicating with the streaming
- * executable. The default if this is not set is \n.
- */
- public void setRecordDelimiter(byte delimiter);
}
interface StreamToPig {
@@ -675, +575 @@
public Tuple deserialize(byte[]) throws IOException;
/**
- * Set the record delimiter to use when reading from the streaming
- * executable. The default if this is not set is \n.
+ * This will be called on the front end during planning and not on the back
+ * end during execution.
+ *
+ * @return the {@link LoadCaster} associated with this object.
+ * @throws IOException if there is an exception during LoadCaster
*/
- public void setRecordDelimiter(byte delimiter);
+ public LoadCaster getLoadCaster() throws IOException;
}
}}}
-
- The default implementation of this would be as suggested above. The syntax for describing how data is (de)serialized would then stay as it currently is, except instead of giving a
+ The default implementation of this would be as suggested above. The syntax for describing how data is (de)serialized would then stay as it currently is, except instead of giving a !StoreFunc the user would specify a !PigToStream, and instead of specifying a !LoadFunc a !StreamToPig.
- !StoreFunc the user would specify a !PigToStream, and instead of specifying a !LoadFunc a !StreamToPig.
Additionally, it has been noted that this change takes away the current optimization of Pig Latin scripts such as the following:
@@ -692, +593 @@
B = stream A through 'mycmd';
store B into 'outfile';
}}}
+ In this case Pig will optimize the query by removing the load function and replacing it with !BinaryStorage, a function which simply passes the data as is to the streaming executable. It does not record or field parsing. Similarly, the store in the above script would be replaced with !BinaryStorage.
+ We have two options to replace this. First, we could say that if a class implementing !PigToStream also implements !InputFormat, then Pig will drop the Load statement and use that !InputFormat directly to load data and then pass the results to the stream. The same would be done with !StreamToPig, !OutputFormat and store. Second, we could create !IdentityLoader and !IdentityStreamToPig functions. !IdentityLoader.getNext would return a tuple that just had one bytearray, which would be the entire record. This would then be a trivial serialization via the default !PigToStream. Similarly !IdentityStreamToPig would take the bytes returned by the stream and put them in a tuple of a single bytearray. The store function would then naturally translate this tuple into the underlying bytes. Functionally these are basically equivalent, since Pig would need to write code similar to the !IdentityLoader etc. for the second case. So I believe the primary difference is in how it is presented to the user not the functionality or code written underneath.
- In this case Pig will optimize the query by removing the load function and replacing it with !BinaryStorage, a function which simply passes the data as is to the streaming
- executable. It does not record or field parsing. Similarly, the store in the above script would be replaced with !BinaryStorage.
- We have two options to replace this. First, we could say that if a class implementing !PigToStream also implements !InputFormat, then Pig will drop the Load statement and use that
- !InputFormat directly to load data and then pass the results to the stream. The same would be done with !StreamToPig, !OutputFormat and store. Second, we could create
- !IdentityLoader and !IdentityStreamToPig functions. !IdentityLoader.getNext would return a tuple that just had one bytearray, which would be the entire record. This would then be a
- trivial serialization via the default !PigToStream. Similarly !IdentityStreamToPig would take the bytes returned by the stream and put them in a tuple of a single bytearray. The
- store function would then naturally translate this tuple into the underlying bytes.
- Functionally these are basically equivalent, since Pig would need to write code similar to the !IdentityLoader etc. for the second case. So I believe the primary difference is in
- how it is presented to the user not the functionality or code written underneath.
-
- Both of these approaches suffer from the problem that they assume !TextInputFormat and !TextOutputFormat. For any other IF/OF it will not be clear how to parse key, value
+ Both of these approaches suffer from the problem that they assume !TextInputFormat and !TextOutputFormat. For any other IF/OF it will not be clear how to parse key, value pairs out of the stream data.
- pairs out of the stream data.
- This optimization represents a fair amount of work. As the current optimization is not documented, it is not clear how many users are using it. Based on that I vote that we
+ This optimization represents a fair amount of work. As the current optimization is not documented, it is not clear how many users are using it. Based on that I vote that we do not implement this optimization until such time as we see a need for it.
- do not implement this optimization until such time as we see a need for it.
=== Remaining Tasks ===
* !BinStorage needs to implement !LoadMetadata's getSchema() to replace current determineSchema()
@@ -718, +609 @@
* !PigDump needs to be ported
* POLoad needs to be ported
* Need to handle passing loadfunc specific info between different instances of loadfunc (Different instances in front end and between front end and back end - we need what is required in PIG-602) (setPartitionFilter() and pushOperators()for example needs this - these methods are called in the front end but the information passed is needed in the backend)
- * For !ResourceSchema to be effectively used for communicating schema, we must fix the two level access issues with schema of bags in current schema before we make these changes, otherwise that same contagion will afflict us here.
+ * For !ResourceSchema to be effectively used for communicating schema, we must fix the two level access issues with schema of bags in current schema before we make these changes, otherwise that same contagion will afflict us here.
- * Input/Output handler code in streaming needs to be ported
+ * Input/Output handler code in streaming needs to be ported
* split by file will have to removed from language
* fix code with FIXME in comment relating to load-store redesign
* Decide on what we should do with !ReversibleLoadFunc and multiquery optimization
- * Need to fix issues with bytesToBag, bytesToTuple and bytesToMap creating complex types with right schema
+ * Need to fix issues with bytesToBag, bytesToTuple and bytesToMap creating complex types with right schema
* Address any '''Open Question'''s in this document
== Changes ==
Sept 23 2009, Gates
+
* Changed setURI to setLocation in !LoadFunc and !StoreFunc. Also changed it to throw IOException in the cases where the passed in location is not valid for this load or store mechanism.
* Changed !LoadSchema to !ResourceSchema and !LoadStatistics to !ResourceStatistics
* Added getPartitionKeys and setPartitionFilter to !LoadMetadata
Sept 25 2009, Gates
+
* Added allFinished call to !StoreFunc
Sept 29 2009, Gates
+
* Added answer for open question 1. Added and answered open questions 2 and 3.
Nov 2 2009, Pradeep Kamath
In !LoadFunc:
- * Added relativeToAbsolutePath() method in !LoadFunc per http://issues.apache.org/jira/browse/PIG-879?focusedCommentId=12768818&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818
+
+ * Added relativeToAbsolutePath() method in !LoadFunc per [[http://issues.apache.org/jira/browse/PIG-879?focusedCommentId=12768818&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#action_12768818|http://issues.apache.org/jira/browse/PIG-879?focusedCommentId=12768818&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818]]
* Changed comments in setLocation regarding the location passed - the location will now be the return value of relativeToAbsolutePath()
* setLocation() now also takes a Job argument since the main purpose of this call is to an opportunity to the !LoadFunc implementation to communicate the input location to underlying !InputFormat. !InputFormat implementations inturn seem to be storing this information inthe Job. For example, !FileInputFormat has the following static method to set the input location: setInputPaths(JobConf conf, String commaSeparatedPaths) ;
* Removed doneReading() method since there is already a RecordReader.close() method which will be called by Hadoop wherein all the functionality that needs to be done on completion of reading can be done.
* All methods now can throw IOException - this keeps the interface more flexible for exception cases
+
In !LoadMetadata:
+
* getSchema(), getStatistics() and getPartitionKeys() methods now take a location and Configuration argument so that the implementation can use that information in returning the information requested.
+
In !StoreFunc:
- * Added relToAbsPathForStoreLocation() method per http://issues.apache.org/jira/browse/PIG-879?focusedCommentId=12768818&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818
+
+ * Added relToAbsPathForStoreLocation() method per [[http://issues.apache.org/jira/browse/PIG-879?focusedCommentId=12768818&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#action_12768818|http://issues.apache.org/jira/browse/PIG-879?focusedCommentId=12768818&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#action_12768818]]
* Methods which did not throw IOException now do so to enable exceptions in implementations
* Removed doneWriting() - same functionality already present in !RecordWriter.close() and !OutputCommitter.commitTask()
* Changed setSchema() to checkSchema since this method is called only to allow StoreFunc to check
@@ -758, +657 @@
Added a new section 'Implementation details and status'
Nov 11, Dmitriy Ryaboy
+
- Minor clarification of meaning of mBytes in !ResourceStatistics
+ . Minor clarification of meaning of mBytes in !ResourceStatistics
- Nov 12 2009, Thejas Nair
- Added sections -
+ Nov 12 2009, Thejas Nair Added sections -
+
* Changes to order-by sampling (!RandomSampler)
* Changes to skew join sampling (!PoissonSampleLoader)
Nov 23 2009, Gates
+
* Added section "Changes to Streaming"
Nov 23 2009, Dmitriy Ryaboy
+
* updated StoreMetadata to match changes made to LoadMetadata
Nov 25 2009, Gates
+
* Updated section on streaming to suggest creating an interface for streaming (de)serializers rather than having only one hardwired option. Also added some thoughts on possible replacements for the current !BinaryStorage/split by file optimization.
+ Dec 16 2009, Richard Ding
+
+ * Updated section on streaming to reflect the current implementation.
+