Fix to correctly load schema for input paths

The map from input path to schema was not set up correctly in the record reader, so the schema wasn't be loaded as a result and instead was null.  This corrects the problem and also adds verification that the schema can be determined.

This also fixes an issue where only schemas for the most recent day of input would be loaded.  This could cause a problems for joins as data for the most recent day may not be available across all inputs.  As a result these missing inputs did not have schemas loaded.

Author: Matt Hayes

Diff for: contrib/hourglass/src/java/datafu/hourglass/avro/AvroMultipleInputsKeyInputFormat.java

@@ -36,13 +36,18 @@
  * container files store only records (not key/value pairs), the value from
  * this InputFormat is a NullWritable.</p>
  */
-public class AvroMultipleInputsKeyInputFormat<T> extends FileInputFormat<AvroKey<T>, NullWritable> {
+public class AvroMultipleInputsKeyInputFormat<T> extends FileInputFormat<AvroKey<T>, NullWritable> 
+{  
   /** {@inheritDoc} */
   @Override
   public RecordReader<AvroKey<T>, NullWritable> createRecordReader(
       InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException 
   {    
-    Schema readerSchema = AvroMultipleInputsUtil.getInputKeySchemaForSplit(context.getConfiguration(), split);    
+    Schema readerSchema = AvroMultipleInputsUtil.getInputKeySchemaForSplit(context.getConfiguration(), split);
+    if (readerSchema == null)
+    {
+      throw new RuntimeException("Could not determine input schema");
+    }
     return new AvroKeyRecordReader<T>(readerSchema);
   }
 }

Diff for: contrib/hourglass/src/java/datafu/hourglass/avro/AvroMultipleInputsUtil.java

@@ -21,6 +21,7 @@
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.log4j.Logger;
 import org.json.JSONException;
 import org.json.JSONObject;
 
@@ -34,6 +35,8 @@
  */
 public class AvroMultipleInputsUtil
 {
+  private static final Logger _log = Logger.getLogger(AvroMultipleInputsUtil.class);
+  
   private static final String CONF_INPUT_KEY_SCHEMA = "avro.schema.multiple.inputs.keys";
 
   /**
@@ -46,6 +49,7 @@ public class AvroMultipleInputsUtil
   public static Schema getInputKeySchemaForSplit(Configuration conf, InputSplit split) 
   {
     String path = ((FileSplit)split).getPath().toString();
+    _log.info("Determining schema for input path " + path);
     JSONObject schemas;
     try
     {
@@ -55,16 +59,18 @@ public static Schema getInputKeySchemaForSplit(Configuration conf, InputSplit sp
     {
       throw new RuntimeException(e1);
     }
-    String schemaString = null;
+    Schema schema = null;
     if (schemas != null)
     {
       for (String key : JSONObject.getNames(schemas))
       {
+        _log.info("Checking against " + key);
         if (path.startsWith(key))
         {
           try
           {
-            schemaString = schemas.getString(key);
+            schema = new Schema.Parser().parse(schemas.getString(key));
+            _log.info("Input schema found: " + schema.toString(true));
             break;
           }
           catch (JSONException e)
@@ -74,7 +80,11 @@ public static Schema getInputKeySchemaForSplit(Configuration conf, InputSplit sp
         }
       }
     }
-    return schemaString != null ? new Schema.Parser().parse(schemaString) : null;
+    if (schema == null)
+    {
+      _log.info("Could not determine input schema");
+    }
+    return schema;
   }
 
   /**

Diff for: contrib/hourglass/src/java/datafu/hourglass/fs/PathUtils.java

@@ -56,7 +56,7 @@ public class PathUtils
   public static final SimpleDateFormat datedPathFormat = new SimpleDateFormat("yyyyMMdd");  
   public static final SimpleDateFormat nestedDatedPathFormat = new SimpleDateFormat("yyyy/MM/dd");  
   private static final Pattern timestampPathPattern = Pattern.compile(".+/(\\d{8})");
-  private static final Pattern dailyPathPattern = Pattern.compile(".+/(\\d{4}/\\d{2}/\\d{2})");
+  private static final Pattern dailyPathPattern = Pattern.compile("(.+)/(\\d{4}/\\d{2}/\\d{2})");
 
   /**
    * Filters out paths starting with "." and "_".
@@ -144,7 +144,7 @@ public static List<DatePath> findNestedDatedPaths(FileSystem fs, Path input) thr
       Matcher matcher = dailyPathPattern.matcher(pathStatus.getPath().toString());
       if (matcher.matches())
       {
-        String datePath = matcher.group(1);
+        String datePath = matcher.group(2);
         Date date;
         try
         {
@@ -299,11 +299,30 @@ public static Date getDateForNestedDatedPath(Path path)
 
     try
     {
-      return PathUtils.nestedDatedPathFormat.parse(matcher.group(1));
+      return PathUtils.nestedDatedPathFormat.parse(matcher.group(2));
     }
     catch (ParseException e)
     {
       throw new RuntimeException("Unexpected input filename: " + path);
     }
   }
+  
+  /**
+   * Gets the root path for a path in the "yyyy/MM/dd" format.  This is part of the path preceding the
+   * "yyyy/MM/dd" portion.
+   * 
+   * @param path
+   * @return
+   */
+  public static Path getNestedPathRoot(Path path)
+  {
+    Matcher matcher = dailyPathPattern.matcher(path.toString());
+    
+    if (!matcher.matches())
+    {
+      throw new RuntimeException("Unexpected input filename: " + path);
+    }
+    
+    return new Path(matcher.group(1));
+  }
 }

Diff for: contrib/hourglass/src/java/datafu/hourglass/jobs/PartitionCollapsingExecutionPlanner.java

@@ -24,11 +24,13 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.SortedMap;
 
 import org.apache.avro.Schema;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.log4j.Logger;
 
 
@@ -80,7 +82,7 @@ public class PartitionCollapsingExecutionPlanner extends ExecutionPlanner
   private List<DatePath> _inputsToProcess = new ArrayList<DatePath>();
   private List<DatePath> _newInputsToProcess = new ArrayList<DatePath>();
   private List<DatePath> _oldInputsToProcess = new ArrayList<DatePath>();
-  private Map<Date,List<DatePath>> _inputsToProcessByDate = new HashMap<Date,List<DatePath>>();
+  private Map<String,String> _latestInputByPath = new HashMap<String,String>();
   private DatePath _previousOutputToProcess;
   private List<Schema> _inputSchemas = new ArrayList<Schema>();
   private Map<String,Schema> _inputSchemasByPath = new HashMap<String,Schema>();
@@ -282,17 +284,17 @@ private void determineNumReducers() throws IOException
    */
   private void determineInputSchemas() throws IOException
   {
-    List<Date> dates = new ArrayList<Date>(_inputsToProcessByDate.keySet());
-    if (dates.size() > 0)
+    if (_latestInputByPath.size() > 0)
     {
-      Collections.sort(dates);
-      Date lastDate = dates.get(dates.size()-1);
-      List<DatePath> lastInputs = _inputsToProcessByDate.get(lastDate);
-      for (DatePath input : lastInputs)
+      _log.info("Determining input schemas");
+      for (Entry<String,String> entry : _latestInputByPath.entrySet())
       {
-        Schema schema = PathUtils.getSchemaFromPath(getFileSystem(),input.getPath());
+        String root = entry.getKey();
+        String input = entry.getValue();
+        _log.info("Loading schema for " + input);
+        Schema schema = PathUtils.getSchemaFromPath(getFileSystem(),new Path(input));
         _inputSchemas.add(schema);
-        _inputSchemasByPath.put(input.getPath().toString(), schema);
+        _inputSchemasByPath.put(root, schema);
       }
     }
   }
@@ -331,7 +333,7 @@ private void determineInputsToProcess() throws IOException
     Calendar cal = Calendar.getInstance(PathUtils.timeZone);    
 
     _inputsToProcess.clear();
-    _inputsToProcessByDate.clear();
+    _latestInputByPath.clear();
     _previousOutputToProcess = null;
 
     DateRange outputDateRange = null;
@@ -359,10 +361,11 @@ private void determineInputsToProcess() throws IOException
           _log.info(String.format("Input: %s",input.getPath()));
           _inputsToProcess.add(input);
           _oldInputsToProcess.add(input);
+          
+          Path root = PathUtils.getNestedPathRoot(input.getPath());
+          _latestInputByPath.put(root.toString(), input.getPath().toString());
         }
-        
-        _inputsToProcessByDate.put(currentDate, inputs);
-                
+                                
         cal.setTime(currentDate);
         cal.add(Calendar.DAY_OF_MONTH, 1);
         currentDate = cal.getTime();
@@ -414,10 +417,11 @@ private void determineInputsToProcess() throws IOException
             _log.info(String.format("Input: %s",input.getPath()));
             _inputsToProcess.add(input);
             _newInputsToProcess.add(input);
+            
+            Path root = PathUtils.getNestedPathRoot(input.getPath());
+            _latestInputByPath.put(root.toString(), input.getPath().toString());
           }
-          
-          _inputsToProcessByDate.put(currentDate, inputs);
-          
+                    
           newDataCount++;
         }
       }

Diff for: contrib/hourglass/src/java/datafu/hourglass/jobs/PartitionPreservingExecutionPlanner.java

@@ -28,9 +28,11 @@
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeSet;
+import java.util.Map.Entry;
 
 import org.apache.avro.Schema;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.log4j.Logger;
 
 import datafu.hourglass.fs.DatePath;
@@ -64,7 +66,7 @@ public class PartitionPreservingExecutionPlanner extends ExecutionPlanner
   private final Logger _log = Logger.getLogger(PartitionPreservingExecutionPlanner.class);
 
   private SortedMap<Date,DatePath> _outputPathsByDate;
-  private Map<Date,List<DatePath>> _inputsToProcessByDate = new HashMap<Date,List<DatePath>>();
+  private Map<String,String> _latestInputByPath = new HashMap<String,String>();
   private List<DatePath> _inputsToProcess = new ArrayList<DatePath>();
   private List<Schema> _inputSchemas = new ArrayList<Schema>();
   private Map<String,Schema> _inputSchemasByPath = new HashMap<String,Schema>();
@@ -208,17 +210,17 @@ private void determineNumReducers() throws IOException
    */
   private void determineInputSchemas() throws IOException
   {
-    List<Date> dates = new ArrayList<Date>(_inputsToProcessByDate.keySet());
-    if (dates.size() > 0)
+    if (_latestInputByPath.size() > 0)
     {
-      Collections.sort(dates);
-      Date lastDate = dates.get(dates.size()-1);
-      List<DatePath> lastInputs = _inputsToProcessByDate.get(lastDate);
-      for (DatePath input : lastInputs)
+      _log.info("Determining input schemas");
+      for (Entry<String,String> entry : _latestInputByPath.entrySet())
       {
-        Schema schema = PathUtils.getSchemaFromPath(getFileSystem(),input.getPath());
+        String root = entry.getKey();
+        String input = entry.getValue();
+        _log.info("Loading schema for " + input);
+        Schema schema = PathUtils.getSchemaFromPath(getFileSystem(),new Path(input));
         _inputSchemas.add(schema);
-        _inputSchemasByPath.put(input.getPath().toString(), schema);
+        _inputSchemasByPath.put(root, schema);
       }
     }
   }
@@ -231,6 +233,7 @@ private void determineInputSchemas() throws IOException
   private void determineInputsToProcess()
   {
     _log.info("Determining inputs to process");
+    _latestInputByPath.clear();
     int newDataCount = 0;
     Calendar cal = Calendar.getInstance(PathUtils.timeZone);
     for (Date currentDate=getDateRange().getBeginDate(); currentDate.compareTo(getDateRange().getEndDate()) <= 0; )
@@ -251,10 +254,11 @@ private void determineInputsToProcess()
           {
             _log.info(String.format("Input: %s",input.getPath()));
             _inputsToProcess.add(input);
+            
+            Path root = PathUtils.getNestedPathRoot(input.getPath());
+            _latestInputByPath.put(root.toString(), input.getPath().toString());
           }
-          
-          _inputsToProcessByDate.put(currentDate, inputs);
-          
+                    
           newDataCount++;
         }
         else

Diff for: contrib/hourglass/test/java/datafu/hourglass/test/PartitionCollapsingJoinTest.java

@@ -173,16 +173,66 @@ public void joinTest() throws IOException, InterruptedException, ClassNotFoundEx
     HashMap<Long,ImpressionClick> counts;
 
     counts = loadOutputCounts("20130317");    
-    checkSize(counts,7);    
+    checkSize(counts,10);    
     checkIdCount(counts,1,1,1);
     checkIdCount(counts,2,1,2);
+    checkIdCount(counts,3,2,0);
     checkIdCount(counts,4,1,1);
     checkIdCount(counts,5,1,1);
     checkIdCount(counts,6,2,1);
+    checkIdCount(counts,7,2,0);
     checkIdCount(counts,8,1,3);
+    checkIdCount(counts,9,1,0);
     checkIdCount(counts,10,1,2);
   }
 
+  @Test
+  public void joinMissingClicksTest() throws IOException, InterruptedException, ClassNotFoundException
+  {    
+    startImpressions(2013, 3, 15);
+    startClicks(2013, 3, 15);
+    impression(1); click(1);
+    impression(2); click(2); click(2);
+    impression(3); impression(3);
+    impression(4); click(4);
+    stopImpressions();
+    stopClicks();
+    
+    startImpressions(2013, 3, 16);
+    startClicks(2013, 3, 16);
+    impression(5); click(5);
+    impression(6); impression(6); click(6);
+    impression(7);
+    impression(7);
+    stopImpressions();
+    stopClicks();
+    
+    startImpressions(2013, 3, 17);
+    impression(8);
+    impression(9);
+    impression(10);
+    stopImpressions();
+    
+    runJob();
+    
+    checkOutputFolderCount(1);
+        
+    HashMap<Long,ImpressionClick> counts;
+    
+    counts = loadOutputCounts("20130317");    
+    checkSize(counts,10);    
+    checkIdCount(counts,1,1,1);
+    checkIdCount(counts,2,1,2);
+    checkIdCount(counts,3,2,0);
+    checkIdCount(counts,4,1,1);
+    checkIdCount(counts,5,1,1);
+    checkIdCount(counts,6,2,1);
+    checkIdCount(counts,7,2,0);
+    checkIdCount(counts,8,1,0);
+    checkIdCount(counts,9,1,0);
+    checkIdCount(counts,10,1,0);
+  }
+  
   private AbstractPartitionCollapsingIncrementalJob runJob() throws IOException, InterruptedException, ClassNotFoundException
   {
     _props = newTestProperties();