title: Spark custom encoder for dataframe
body: <p>I know about <a href="https://stackoverflow.com/questions/36648128/how-to-store-custom-objects-in-a-dataset">How to store custom objects in Dataset?</a> but still, it is not really clear for me how to build this <strong>custom encoder which properly serializes to multiple fields.</strong> Manually, I created some functions <a href="https://github.com/geoHeil/geoSparkScalaSample/blob/master/src/main/scala/myOrg/GeoSpark.scala#L122-L154" rel="nofollow noreferrer">https://github.com/geoHeil/geoSparkScalaSample/blob/master/src/main/scala/myOrg/GeoSpark.scala#L122-L154</a> wich map a Polygon back and forth between <code>Dataset - RDD - Dataset</code> by mapping the objects to primitive types spark can handle i.e. a tuple <code>(String, Int)</code>(edit: full code below).</p>
  
  <p>For example, to go from the Polygon Object to a tuple of <code>(String, Int)</code> I use the following</p>
  
  <pre><code>def writeSerializableWKT(iterator: Iterator[AnyRef]): Iterator[(String, Int)] = {
      val writer = new WKTWriter()
      iterator.flatMap(cur =&gt; {
        val cPoly = cur.asInstanceOf[Polygon]
        // TODO is it efficient to create this collection? Is this a proper iterator 2 iterator transformation?
        List(((writer.write(cPoly), cPoly.getUserData.asInstanceOf[Int])).iterator
      })
    }
   def createSpatialRDDFromLinestringDataSet(geoDataset: Dataset[WKTGeometryWithPayload]): RDD[Polygon] = {
      geoDataset.rdd.mapPartitions(iterator =&gt; {
        val reader = new WKTReader()
        iterator.flatMap(cur =&gt; {
          try {
            reader.read(cur.lineString) match {
              case p: Polygon =&gt; {
                val polygon = p.asInstanceOf[Polygon]
                polygon.setUserData(cur.payload)
                List(polygon).iterator
              }
              case _ =&gt; throw new NotImplementedError("Multipolygon or others not supported")
            }
          } catch {
            case e: ParseException =&gt;
              logger.error("Could not parse")
              logger.error(e.getCause)
              logger.error(e.getMessage)
              None
          }
        })
      })
    }
  </code></pre>
  
  <p>I noticed that already I start to do a lot of work twice (see the link to both methods). Now wanting to be able to handle </p>
  
  <p><a href="https://github.com/geoHeil/geoSparkScalaSample/blob/master/src/main/scala" rel="nofollow noreferrer">https://github.com/geoHeil/geoSparkScalaSample/blob/master/src/main/scala</a> (full code below)</p>
  
  <pre><code>/myOrg/GeoSpark.scala#L82-L84
   val joinResult = JoinQuery.SpatialJoinQuery(objectRDD, minimalPolygonCustom, true)
    //  joinResult.map()
    val joinResultCounted = JoinQuery.SpatialJoinQueryCountByKey(objectRDD, minimalPolygonCustom, true)
  </code></pre>
  
  <p>which is a <code>PairRDD[Polygon, HashSet[Polygon]]</code>, or respectively <code>PairRDD[Polygon, Int]</code> how would I need to specify my functions as an Encoder in order to not solve the same problem 2 more times?</p>