创建1.0.5分支，修改时间字段不支持问题

build.sbt

@@ -32,7 +32,8 @@ libraryDependencies ++= Seq(
 libraryDependencies ++= Seq(
   "org.apache.hadoop" % "hadoop-client" % testHadoopVersion.value % "test",
   "org.apache.spark" %% "spark-core" % testSparkVersion.value % "test" exclude("org.apache.hadoop", "hadoop-client"),
-  "org.apache.spark" %% "spark-sql" % testSparkVersion.value % "test" exclude("org.apache.hadoop", "hadoop-client")
+  "org.apache.spark" %% "spark-sql" % testSparkVersion.value % "test" exclude("org.apache.hadoop", "hadoop-client"),
+  "org.apache.hadoop" % "hadoop-client" % testHadoopVersion.value % "provided"
 )
 
 // This is necessary because of how we explicitly specify Spark dependencies

project/build.properties

@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-sbt.version=0.13.6
+sbt.version=0.13.13

src/main/scala/magellan/io/ShapeReader.scala

@@ -41,8 +41,8 @@ class NullShapeReader extends ShapeReader {
 private[magellan] class PointReader extends ShapeReader {
 
   override def readFields(dataInput: DataInput): Shape = {
-    val x = EndianUtils.swapDouble(dataInput.readDouble())
-    val y = EndianUtils.swapDouble(dataInput.readDouble())
+    val x = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
+    val y = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
     Point(x, y)
   }
 
@@ -52,10 +52,10 @@ private[magellan] class PolygonReader extends ShapeReader {
 
   override def readFields(dataInput: DataInput): Polygon = {
     // extract bounding box.
-    val xmin = EndianUtils.swapDouble(dataInput.readDouble())
-    val ymin = EndianUtils.swapDouble(dataInput.readDouble())
-    val xmax = EndianUtils.swapDouble(dataInput.readDouble())
-    val ymax = EndianUtils.swapDouble(dataInput.readDouble())
+    val xmin = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
+    val ymin = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
+    val xmax = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
+    val ymax = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
 
     val box = BoundingBox(xmin, ymin, xmax, ymax)
 
@@ -81,8 +81,8 @@ private[magellan] class PolygonReader extends ShapeReader {
     val ycoordinates = Array.fill(numPoints)(0.0)
 
     for (index <- 0 until numPoints) {
-      val x = Double.longBitsToDouble(Long.reverseBytes(dataInput.readLong()))
-      val y = Double.longBitsToDouble(Long.reverseBytes(dataInput.readLong()))
+      val x = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
+      val y = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
       xcoordinates.update(index, x)
       ycoordinates.update(index, y)
     }
@@ -96,7 +96,7 @@ private[magellan] class PolyLineReader extends ShapeReader {
 
   def extract(dataInput: DataInput): (Array[Int], Array[Point]) = {
     // extract bounding box.
-    (0 until 4).foreach { _ => EndianUtils.swapDouble(dataInput.readDouble())}
+    (0 until 4).foreach { _ => Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))}
 
     // numRings
     val numRings = EndianUtils.swapInteger(dataInput.readInt())
@@ -119,8 +119,8 @@ private[magellan] class PolyLineReader extends ShapeReader {
     val points = ArrayBuffer[Point]()
     for (_ <- 0 until numPoints) {
       points.+= {
-        val x = EndianUtils.swapDouble(dataInput.readDouble())
-        val y = EndianUtils.swapDouble(dataInput.readDouble())
+        val x = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
+        val y = Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong()))
         Point(x, y)
       }
     }
@@ -139,7 +139,7 @@ private[magellan] class PolyLineZReader extends PolyLineReader {
     val (indices, points) = extract(dataInput)
     // throw away the Z and M values
     val size = points.length
-    (0 until (4 + 2 * size)).foreach(_ => dataInput.readDouble())
+    (0 until (4 + 2 * size)).foreach(_ => Double.longBitsToDouble(EndianUtils.swapLong(dataInput.readLong())))
     if(indices.size != points.size)
       PolyLine( new Array[Int](points.size), points)
     else

src/main/scala/magellan/mapreduce/DBReader.scala

@@ -17,16 +17,15 @@
 package magellan.mapreduce
 
 import java.io.DataInputStream
+import java.text.SimpleDateFormat
 
-import scala.collection.mutable.ListBuffer
-
+import magellan.io.ShapeKey
 import org.apache.commons.io.EndianUtils
-import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.io._
 import org.apache.hadoop.mapreduce.lib.input.FileSplit
 import org.apache.hadoop.mapreduce.{InputSplit, RecordReader, TaskAttemptContext}
 
-import magellan.io.ShapeKey
+import scala.collection.mutable.ListBuffer
 
 private[magellan] class DBReader extends RecordReader[ShapeKey, MapWritable] {
 
@@ -73,16 +72,38 @@ private[magellan] class DBReader extends RecordReader[ShapeKey, MapWritable] {
         case 'C' => {
           val b = Array.fill[Byte](length)(0)
           dis.readFully(b)
+          val value = new String(b).trim
           val fld = new Text()
-          fld.append(b, 0, length)
+          fld.clear()
+          fld.set(value)
           fld
         }
         case 'N' | 'F' => {
           val b = Array.fill[Byte](length)(0)
           dis.readFully(b)
+          var value = new String(b).trim
+          val index = value.indexOf(0)
+          if (index != -1) {
+            value = value.substring(0, index)
+          }
           val fld = new Text()
           fld.clear()
-          fld.set(new String(b))
+          fld.set(value)
+          fld
+        }
+        case 'D' => {
+          val b = Array.fill[Byte](length)(0)
+          dis.readFully(b, 0, length)
+          val fld = new Text()
+          fld.clear()
+          if (b(0) != '0') {
+            val value = new String(b).trim
+            if (!value.isEmpty) {
+              val format = new SimpleDateFormat(if (length == 8 ) "yyyyMMdd" else "yyyyMMddHHmmss")
+              val date = format.parse(value)
+              fld.set(date.toString)
+            }
+          }
           fld
         }
 
@@ -115,6 +136,9 @@ private[magellan] class DBReader extends RecordReader[ShapeKey, MapWritable] {
     val date = Array.fill[Byte](3)(0)
     dis.read(date)
     numRecords = EndianUtils.swapInteger(dis.readInt())
+    if (numRecords == 0) {
+      return
+    }
     val numBytesInHeader = EndianUtils.swapShort(dis.readShort())
     numBytesInRecord = EndianUtils.swapShort(dis.readShort())
     // skip the next 20 bytes

src/main/scala/magellan/mapreduce/ShapeInputFormat.scala

@@ -65,6 +65,11 @@ private[magellan] class ShapeInputFormat
         splits.+= (makeSplit(path, 0, length, blkLocations(blkIndex).getHosts,
           blkLocations(blkIndex).getCachedHosts))
       } else {
+        val s = splitInfos(key).toSeq
+        val blkIndex = getBlockIndex(blkLocations, s(0))
+        splits.+=(makeSplit(path, s(0), length - s(0), blkLocations(blkIndex).getHosts,
+          blkLocations(blkIndex).getCachedHosts))
+        /*
         val s = splitInfos(key).toSeq
         val start = s
         val end = s.drop(1) ++ Seq(length)
@@ -73,6 +78,7 @@ private[magellan] class ShapeInputFormat
           splits.+=(makeSplit(path, startOffset, endOffset - startOffset, blkLocations(blkIndex).getHosts,
             blkLocations(blkIndex).getCachedHosts))
         }
+        */
       }
     }
     sw.stop