Native python converters for HBaseResult and others

build.sbt

@@ -1,16 +1,16 @@
 lazy val root = (project in file(".")).
   settings(
     name := "spark_hbase",
-    version := "1.0",
+    version := "1.1",
     scalaVersion := "2.10.5",
     sparkVersion := "1.3.0"
   )
 
 libraryDependencies ++= Seq(
-  "org.apache.hbase" % "hbase" % "0.98.8-hadoop2",
-  "org.apache.hbase" % "hbase-client" % "0.98.8-hadoop2",
-  "org.apache.hbase" % "hbase-common" % "0.98.8-hadoop2",
-  "org.apache.hbase" % "hbase-server" % "0.98.8-hadoop2"
+  "org.apache.hbase" % "hbase" % "1.0.0",
+  "org.apache.hbase" % "hbase-client" % "1.0.0",
+  "org.apache.hbase" % "hbase-common" % "1.0.0",
+  "org.apache.hbase" % "hbase-server" % "1.0.0"
 )
 
 mergeStrategy in assembly := {

src/main/scala/examples/pythonConverters.scala

@@ -31,7 +31,7 @@ import org.apache.hadoop.hbase.CellUtil
 /**
  * Implementation of [[org.apache.spark.api.python.Converter]] that converts all 
  * the records in an HBase Result to a String. In the String, it contains row, column,
- * qualifier, timesstamp, type and value
+ * qualifier, timestamp, type and value
  */
 
 class HBaseResultToStringConverter extends Converter[Any, String]{
@@ -54,11 +54,101 @@ class HBaseResultToStringConverter extends Converter[Any, String]{
   }
 }
 
+/**
+ * Implementation of [[org.apache.spark.api.python.Converter]] that converts all 
+ * the records in an HBase Result to a String. In the String, it contains row, column,
+ * qualifier, timestamp, type and value, all packed as JSON
+ */
+class HBaseResultToJSONConverter extends Converter[Any, String]{
+  override def convert(obj: Any): String = {
+    import collection.JavaConverters._
+    val result = obj.asInstanceOf[Result]
+    val output = result.listCells.asScala.map(cell =>
+        Map(
+          "row" -> Bytes.toStringBinary(CellUtil.cloneFamily(cell)),
+          "columnFamily" -> Bytes.toStringBinary(CellUtil.cloneFamily(cell)),
+          "qualifier" -> Bytes.toStringBinary(CellUtil.cloneQualifier(cell)),
+          "timestamp" -> cell.getTimestamp.toString,
+          "type" -> Type.codeToType(cell.getTypeByte).toString,
+          "value" -> Bytes.toStringBinary(CellUtil.cloneValue(cell))
+        )
+      )
+    // output is a JSON array of objects (maps)
+    "[" + output.map(JSONObject(_).toString()).mkString(", ") + "]"
+  }
+}
+
+/**
+ * Implementation of [[org.apache.spark.api.python.Converter]] that converts all 
+ * the records in an HBase Result into a list of maps, each containing row, column,
+ * qualifier, timestamp, type and value.  Values are returned as raw bytes rather than as
+ * printable versions, and thus may require unpacking.
+ */
+class HBaseResultToListConverter extends Converter[Any, java.util.List[java.util.Map[String, String]]]{
+  override def convert(obj: Any): java.util.List[java.util.Map[String, String]] = {
+    import collection.JavaConverters._
+    val result = obj.asInstanceOf[Result]
+    val output = result.listCells.asScala.map(cell =>
+        new java.util.HashMap[String, String](Map(
+          "row" -> Bytes.toString(CellUtil.cloneFamily(cell)),
+          "columnFamily" -> Bytes.toString(CellUtil.cloneFamily(cell)),
+          "qualifier" -> Bytes.toString(CellUtil.cloneQualifier(cell)),
+          "timestamp" -> cell.getTimestamp.toString,
+          "type" -> Type.codeToType(cell.getTypeByte).toString,
+          "value" -> Bytes.toString(CellUtil.cloneValue(cell))
+        ).asJava)
+      )
+    new java.util.ArrayList[java.util.Map[String, String]](output.asJava)
+  }
+}
+
+/**
+ * Implementation of [[org.apache.spark.api.python.Converter]] that converts all 
+ * the records in an HBase Result into a Map of "columnFamily:column"->"value" consistent
+ * with a Python dict.
+ * 
+ * Only works with 1 version max (possibly latest if multiple are returned?).  
+ * Ser/deser as python dict naturally, via HashMap.
+ *
+ * Values are returned as raw bytes rather than as printable versions, and thus
+ * may require unpacking.
+ */
+class HBaseResultToMapConverter extends Converter[Any, java.util.Map[String, String]]{
+  override def convert(obj: Any): java.util.Map[String, String] = {
+
+    import collection.JavaConverters._
+    val result = obj.asInstanceOf[Result]
+
+    val my_map = result.listCells.asScala.map(
+        cell => (
+            Bytes.toString(CellUtil.cloneFamily(cell)) + ":" +
+            Bytes.toString(CellUtil.cloneQualifier(cell)),
+            Bytes.toString(CellUtil.cloneValue(cell)))
+    ).toMap.asJava
+
+    // nb: java.util.Map and scala map do not serialize into python??
+    new java.util.HashMap[String, String](my_map)
+  }
+}
+
+/**
+ * Returns the most recent cell timestamp on the row as a long integer, useful for 
+ * operations that require syncing to data changes.
+ */
+class MaxHBaseTimestamp extends Converter[Any, Long]{
+  override def convert(obj: Any): Long = {
+    import collection.JavaConverters._
+    val result = obj.asInstanceOf[Result]
+    result.listCells.asScala.map(cell =>
+        cell.getTimestamp
+      ).max
+  }
+}
+
 /**
  * Implementation of [[org.apache.spark.api.python.Converter]] that converts an
  * ImmutableBytesWritable to a String
  */
-
 class ImmutableBytesWritableToStringConverter extends Converter[Any, String] {
   override def convert(obj: Any): String = {
     val key = obj.asInstanceOf[ImmutableBytesWritable]