Merge pull request #12 from intenthq/spark-2

Update to spark 2.0

Author: cdagraca

README.md

@@ -12,29 +12,29 @@ The following top level dependencies are published in Maven central:
 
 **Thrift support**:
 ```
-"com.intenthq.pucket" %% "pucket-thrift" % "1.2.0"
+"com.intenthq.pucket" %% "pucket-thrift" % "1.3.0"
 ```
 
 **Avro support**:
 ```
-"com.intenthq.pucket" %% "pucket-avro" % "1.2.0"
+"com.intenthq.pucket" %% "pucket-avro" % "1.3.0"
 ```
 
 **Spark connectors**:
 ```
-"com.intenthq.pucket" %% "pucket-spark" % "1.2.0"
+"com.intenthq.pucket" %% "pucket-spark" % "1.3.0"
 ```
 
 **MapReduce integration**:
 ```
-"com.intenthq.pucket" %% "pucket-mapreduce" % "1.2.0"
+"com.intenthq.pucket" %% "pucket-mapreduce" % "1.3.0"
 ```
 
 These dependencies should be combined depending on your usages; for example if you use Thrift and Spark then use the following:
 
 ```
-"com.intenthq.pucket" %% "pucket-thrift" % "1.2.0"
-"com.intenthq.pucket" %% "pucket-spark" % "1.2.0"
+"com.intenthq.pucket" %% "pucket-thrift" % "1.3.0"
+"com.intenthq.pucket" %% "pucket-spark" % "1.3.0"
 ```
 
 

avro/src/main/scala/com/intenthq/pucket/avro/AvroPucketDescriptor.scala

@@ -42,26 +42,24 @@ object AvroPucketDescriptor extends PucketDescriptorCompanion {
 
   val avroSchemaKey = "avroSchema"
 
-  private def validate[T <: HigherType](descriptorString: String): Throwable \/ (Map[String, String], Schema, CompressionCodecName, Option[PucketPartitioner[T]]) =
+  private def validate[T <: HigherType](descriptorString: String): Throwable \/ (Schema, CompressionCodecName, Option[PucketPartitioner[T]]) =
     for {
       underlying <- parseDescriptor[T](descriptorString)
       schema <- extractValue(underlying._1, avroSchemaKey)
-      parsedSchema <- parse(schema)
       avroSchema <- \/.fromTryCatchNonFatal(new Schema.Parser().parse(schema))
-    } yield (parsedSchema, avroSchema, underlying._2, underlying._3)
+    } yield (avroSchema, underlying._2, underlying._3)
 
   /** @inheritdoc */
   override def apply[T <: IndexedRecord](expectedSchema: Schema, descriptorString: String): Throwable \/ AvroPucketDescriptor[T] =
     for {
       underlying <- validate[T](descriptorString)
-      expectedSchemaJson <- parse(expectedSchema.toString)
-      _ <- if (underlying._2 == expectedSchema) ().right
+      _ <- if (underlying._1 == expectedSchema) ().right
            else new RuntimeException("Found schema does not match expected").left
-    } yield AvroPucketDescriptor[T](underlying._2, underlying._3, underlying._4)
+    } yield AvroPucketDescriptor[T](underlying._1, underlying._2, underlying._3)
 
   /** @inheritdoc */
   override def apply[T <: HigherType](descriptorString: String): Throwable \/ AvroPucketDescriptor[T] =
-    validate[T](descriptorString).map(underlying => AvroPucketDescriptor[T](underlying._2, underlying._3, underlying._4))
+    validate[T](descriptorString).map(underlying => AvroPucketDescriptor[T](underlying._1, underlying._2, underlying._3))
 }
 
 

avro/src/test/scala/com/intenthq/pucket/avro/AvroPucketDescriptorSpec.scala

@@ -31,15 +31,11 @@ class AvroPucketDescriptorSpec extends Specification with DisjunctionMatchers wi
 
   def badString =
     Prop.forAll(randomString) { s =>
-      AvroPucketDescriptor[AvroTest](AvroTest.getClassSchema, s) must be_-\/[Throwable].like {
-        case a => a must beAnInstanceOf[jodd.json.JsonException]
-      }
+      AvroPucketDescriptor[AvroTest](AvroTest.getClassSchema, s) must be_-\/[Throwable]
     }
 
   def badSchema =
-    AvroPucketDescriptor[AvroTest](AvroTest.getClassSchema, badSchemaString) must be_-\/.like {
-      case a => a must beAnInstanceOf[jodd.json.JsonException]
-    }
+    AvroPucketDescriptor[AvroTest](AvroTest.getClassSchema, badSchemaString) must be_-\/[Throwable]
 
   def schemaClassIncorrect =
     AvroPucketDescriptor[AvroTest](AvroTest.getClassSchema, incorrectSchema) must be_-\/.like {

build.sbt

@@ -4,8 +4,9 @@ import com.typesafe.sbt.SbtGit.{GitKeys => git}
 
 val specs2Ver = "3.6.4"
 val parquetVer = "1.8.1"
-val hadoopVer = "2.7.1"
-val sparkVer = "1.6.1"
+val hadoopVer = "2.7.3"
+val sparkVer = "2.0.0"
+val circeVersion = "0.5.1"
 
 val pomInfo = (
   <url>https://github.com/intenthq/pucket</url>
@@ -32,7 +33,7 @@ def excludeServlet(deps: Seq[ModuleID]) = deps.map(_.exclude("javax.servlet", "s
 
 lazy val commonSettings = Seq(
   organization := "com.intenthq.pucket",
-  version := "1.2.1",
+  version := "1.3.0",
   scalaVersion := "2.11.8",
   publishArtifact in Test := false,
   pomIncludeRepository := { _ => false },
@@ -48,13 +49,17 @@ lazy val commonSettings = Seq(
   autoAPIMappings := true,
   libraryDependencies ++= excludeServlet(Seq(
     "org.scalaz" %% "scalaz-core" % "7.1.3",
-    "org.jodd" % "jodd-json" % "3.7",
     "org.mortbay.jetty" % "servlet-api" % "3.0.20100224" % "provided",
     "org.apache.hadoop" % "hadoop-common" % hadoopVer % "provided",
     "org.apache.hadoop" % "hadoop-mapreduce-client-core" % hadoopVer % "provided",
     "org.apache.parquet" % "parquet-column" % parquetVer,
     "org.apache.parquet" % "parquet-hadoop" % parquetVer
   )),
+  libraryDependencies ++= Seq(
+    "io.circe" %% "circe-core",
+    "io.circe" %% "circe-generic",
+    "io.circe" %% "circe-parser"
+  ).map(_ % circeVersion),
   dependencyOverrides += "org.slf4j" % "slf4j-log4j12" % "1.7.12",
   dependencyOverrides += "org.slf4j" % "slf4j-api" % "1.7.12",
   resolvers ++= Seq(
@@ -145,6 +150,7 @@ lazy val spark = (project in file("spark")).
         ExclusionRule(organization = "org.slf4j"),
         ExclusionRule(organization = "log4j"),
         ExclusionRule(organization = "org.scala-lang"),
+        ExclusionRule(organization = "org.scalatest"),
         ExclusionRule(organization = "javax.servlet", name = "servlet-api")
        )
     ))

core/src/main/scala/com/intenthq/pucket/PucketDescriptor.scala

@@ -1,12 +1,12 @@
 package com.intenthq.pucket
 
 import com.intenthq.pucket.util.PucketPartitioner
-import jodd.json.{JsonParser, JsonSerializer}
+import io.circe.parser._
+import io.circe.syntax._
 import org.apache.hadoop.fs.Path
 import org.apache.parquet.hadoop.api.ReadSupport
 import org.apache.parquet.hadoop.metadata.CompressionCodecName
 
-import scala.collection.JavaConverters._
 import scalaz.\/
 import scalaz.syntax.either._
 
@@ -16,8 +16,6 @@ import scalaz.syntax.either._
   * @tparam T the type of data being stored in the pucket
   */
 trait PucketDescriptor[T] {
-  val jsonSerializer = new JsonSerializer()
-
   import PucketDescriptor._
   /** Parquet compression codec */
   def compression: CompressionCodecName
@@ -42,7 +40,7 @@ trait PucketDescriptor[T] {
    *
    * @return JSON string of descriptor
    */
-  override def toString: String = jsonSerializer.serialize(json.asJava)
+  override def toString: String = json.asJson.spaces2
 }
 
 /** Trait for implementations of the pucket descriptor companion object */
@@ -108,5 +106,5 @@ object PucketDescriptor {
       fold[Throwable \/ String](new RuntimeException(s"Could not find $key in descriptor").left)(_.right)
 
   def parse(json: String): Throwable \/ Map[String, String] =
-    \/.fromTryCatchNonFatal(new JsonParser().parse(json).asInstanceOf[java.util.Map[String, String]].asScala.toMap)
+    decode[Map[String, String]](json).fold(_.getCause.left, _.right)
 }

spark/src/main/scala/com/intenthq/pucket/spark/PucketSparkAdapter.scala

@@ -10,7 +10,6 @@ import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 
 import scala.collection.JavaConversions._
-
 import scala.reflect.ClassTag
 /** Provides classes to extend Spark's RDD so that puckets can be read and written with spark */
 object PucketSparkAdapter {

thrift/src/test/scala/com/intenthq/pucket/thrift/ThriftPucketDescriptorSpec.scala

@@ -30,9 +30,7 @@ class ThriftPucketDescriptorSpec extends Specification with DisjunctionMatchers
 
   def badString =
     Prop.forAll(randomString) { s =>
-      ThriftPucketDescriptor[ThriftTest](classOf[ThriftTest], s) must be_-\/[Throwable].like {
-        case a => a must beAnInstanceOf[jodd.json.JsonException]
-      }
+      ThriftPucketDescriptor[ThriftTest](classOf[ThriftTest], s) must be_-\/[Throwable]
     }
 
   def schemaClassNotFound =