diff --git a/build.sbt b/build.sbt index 0c68167..6f1ce11 100644 --- a/build.sbt +++ b/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 := { diff --git a/src/main/scala/examples/pythonConverters.scala b/src/main/scala/examples/pythonConverters.scala index 9a3cf0c..31df49e 100644 --- a/src/main/scala/examples/pythonConverters.scala +++ b/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]