Checking size in tap first and then calculate by ourselves

twitter · Feb 22, 2017 · 5e41f89 · 5e41f89
1 parent 6f11ced
commit 5e41f89
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 4 deletions.
diff --git a/scalding-commons/src/main/java/com/twitter/scalding/commons/tap/VersionedTap.java b/scalding-commons/src/main/java/com/twitter/scalding/commons/tap/VersionedTap.java
@@ -4,6 +4,7 @@
 
 import com.twitter.scalding.commons.datastores.VersionedStore;
 
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.FileInputFormat;
@@ -90,6 +91,15 @@ public String getSinkPath(JobConf conf) {
     }
   }
 
+  @Override
+  public long getSize(JobConf conf) throws IOException {
+    Path path = new Path(getSourcePath(conf));
+
+    return getFileSystem(conf)
+        .getContentSummary(path)
+        .getLength();
+  }
+
   @Override
   public void sourceConfInit(FlowProcess<JobConf> process, JobConf conf) {
     super.sourceConfInit(process, conf);

diff --git a/scalding-core/src/main/scala/com/twitter/scalding/reducer_estimation/Common.scala b/scalding-core/src/main/scala/com/twitter/scalding/reducer_estimation/Common.scala
@@ -59,10 +59,15 @@ object Common {
    * pattern in its path, so we must be ready to handle that case.
    */
   def size(f: Hfs, conf: JobConf): Long = {
-    val fs = f.getPath.getFileSystem(conf)
-    fs.globStatus(f.getPath)
-      .map{ s => fs.getContentSummary(s.getPath).getLength }
-      .sum
+    val sizeInBytes = f.getSize(conf)
+    if (sizeInBytes > 0) {
+      sizeInBytes
+    } else {
+      val fs = f.getPath.getFileSystem(conf)
+      fs.globStatus(f.getPath)
+        .map{ s => fs.getContentSummary(s.getPath).getLength }
+        .sum
+    }
   }
 
   def inputSizes(step: FlowStep[JobConf]): Seq[(String, Long)] = {