Make sure getBlockLocations uses offset and length to find the blocks…

… on HDFS
apache · Oct 25, 2014 · 5cce16f · 5cce16f
1 parent eadde56
commit 5cce16f
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/rdd/HDFSBackedBlockRDD.scala b/streaming/src/main/scala/org/apache/spark/streaming/rdd/HDFSBackedBlockRDD.scala
@@ -86,7 +86,8 @@ class HDFSBackedBlockRDD[T: ClassTag](
     val partition = split.asInstanceOf[HDFSBackedBlockRDDPartition]
     val locations = getBlockIdLocations()
     locations.getOrElse(partition.blockId,
-      HdfsUtils.getBlockLocations(partition.segment.path, hadoopConfiguration)
+      HdfsUtils.getBlockLocations(partition.segment.path, partition.segment.offset,
+        partition.segment.length, hadoopConfiguration)
         .getOrElse(new Array[String](0)).toSeq)
   }
 }
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/util/HdfsUtils.scala b/streaming/src/main/scala/org/apache/spark/streaming/util/HdfsUtils.scala
@@ -52,11 +52,12 @@ private[streaming] object HdfsUtils {
     }
   }
 
-  def getBlockLocations(path: String, conf: Configuration): Option[Array[String]] = {
+  def getBlockLocations(path: String, offset: Long, length: Long, conf: Configuration):
+  Option[Array[String]] = {
     val dfsPath = new Path(path)
     val dfs = getFileSystemForPath(dfsPath, conf)
     val fileStatus = dfs.getFileStatus(dfsPath)
-    val blockLocs = Option(dfs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen))
+    val blockLocs = Option(dfs.getFileBlockLocations(fileStatus, offset, length))
     blockLocs.map(_.flatMap(_.getHosts))
   }