implement the FlatMap typeclass on Job. This allows Job to be used as a Kleisli/Reader

build.sbt

@@ -1,10 +1,14 @@
 val sparkVersion = "2.3.1"
 val catsCoreVersion = "1.4.0"
+val catsLawVersion = "1.1.0"
 val catsEffectVersion = "1.0.0"
 val catsMtlVersion = "0.3.0"
 val scalatest = "3.0.3"
 val shapeless = "2.3.2"
 val scalacheck = "1.13.5"
+val scalacheckShaplessVersion = "1.1.6"
+val magnoliaVersion = "0.6.1"
+val catsTestkitVersion = "1.1.0"
 
 lazy val root = Project("frameless", file("." + "frameless")).in(file("."))
   .aggregate(core, cats, dataset, ml, docs)
@@ -32,6 +36,9 @@ lazy val cats = project
     "org.typelevel"    %% "cats-effect"    % catsEffectVersion,
     "org.typelevel"    %% "cats-mtl-core"  % catsMtlVersion,
     "org.typelevel"    %% "alleycats-core" % catsCoreVersion,
+    "com.propensive"   %% "magnolia"       % magnoliaVersion    % Test,
+    "org.typelevel"    %% "cats-testkit"   % catsTestkitVersion % Test, //or `cats-testkit` if you are using ScalaTest
+    "com.github.alexarchambault"           %% "scalacheck-shapeless_1.13" % scalacheckShaplessVersion % Test,
     "org.apache.spark" %% "spark-core"     % sparkVersion % "provided",
     "org.apache.spark" %% "spark-sql"      % sparkVersion % "provided"))
   .dependsOn(dataset % "test->test;compile->compile")

cats/src/main/scala/frameless/cats/implicits.scala

@@ -8,6 +8,7 @@ import alleycats.Empty
 
 import scala.reflect.ClassTag
 import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.SparkSession
 
 object implicits extends FramelessSyntax with SparkDelayInstances {
   implicit class rddOps[A: ClassTag](lhs: RDD[A]) {
@@ -72,3 +73,35 @@ object outer {
         }
     }
 }
+
+object jobber {
+
+  implicit def flatMapJob: FlatMap[Job] =
+    new FlatMap[Job] {
+
+      override def flatMap[A, B](j: Job[A])(f: A => Job[B]): Job[B] = j.flatMap(f)
+
+      override def tailRecM[A, B](a: A)(f: A => Job[Either[A, B]]): Job[B] = {
+        val j = f(a)
+        for {
+          either_ab <- j
+          b <- either_ab match {
+            case Left(la) => tailRecM(la)(f)
+            case Right(_) => j.map(_.right.get)
+          }
+        } yield b
+      }
+
+      override def map[A, B](fa: Job[A])(f: A => B): Job[B] = fa.map(f)
+    }
+  implicit def eqJob[A: Eq]: Eq[Job[A]] = Eq.fromUniversalEquals
+
+  implicit def monadJob(implicit spark: SparkSession): Monad[Job] = new Monad[Job]{
+
+    override def pure[A](x: A): Job[A] =  Job(x)
+
+    override def flatMap[A, B](fa: Job[A])(f: A => Job[B]): Job[B] = flatMapJob.flatMap(fa)(f)
+
+    override def tailRecM[A, B](a: A)(f: A => Job[Either[A, B]]): Job[B] = flatMapJob.tailRecM(a)(f)
+  }
+}

cats/src/test/scala/frameless/cats/LawTests.scala

@@ -0,0 +1,23 @@
+package frameless.cats
+
+
+import cats.laws.discipline.{FunctorTests, MonadTests}
+import cats.tests.CatsSuite
+import frameless.{Job, TypedDatasetSuite}
+import org.scalacheck.ScalacheckShapeless._
+import jobber._
+import org.scalacheck.{Arbitrary, Gen}
+
+
+
+class LawTests extends  TypedDatasetSuite  with CatsSuite {
+
+
+  implicit def genJob[A: Arbitrary]: Gen[Job[A]] = for {
+   a <- Arbitrary.arbitrary[A]
+  } yield Job(a)
+
+  implicit def arbJob[A: Arbitrary]: Arbitrary[Job[A]] = Arbitrary[Job[A]](genJob)
+    checkAll("Job.MonadLaws", MonadTests[Job].monad[Int,String,Map[String,Int]])
+    checkAll("Job.FunctorLaws", FunctorTests[Job].functor[Int,String,Map[String,Int]])
+}