title: Spark Structured Streaming MemoryStream + Row + Encoders issue
body: <p>I am trying to run some tests on my local machine with spark structured streaming. </p>
  
  <p>In batch mode here is the Row that i am dealing with: </p>
  
  <pre><code>val recordSchema = StructType(List(StructField("Record", MapType(StringType, StringType), false)))
  val rows         = List(
      Row(
        Map("ID" -&gt; "1",
          "STRUCTUREID" -&gt; "MFCD00869853",
          "MOLFILE" -&gt; "The MOL Data",
          "MOLWEIGHT" -&gt; "803.482",
          "FORMULA" -&gt; "C44H69NO12",
          "NAME" -&gt; "Tacrolimus",
          "HASH" -&gt; "52b966c551cfe0fa7d526bac16abcb7be8b8867d",
          "SMILES" -&gt; """[H][C@]12O[C@](O)([C@H](C)C[C@@H]1OC)""",
          "METABOLISM" -&gt; "The metabolism 500"
         )),
      Row(
        Map("ID" -&gt; "2",
          "STRUCTUREID" -&gt; "MFCD00869854",
          "MOLFILE" -&gt; "The MOL Data",
          "MOLWEIGHT" -&gt; "603.482",
          "FORMULA" -&gt; "",
          "NAME" -&gt; "Tacrolimus2",
          "HASH" -&gt; "52b966c551cfe0fa7d526bac16abcb7be8b8867d",
          "SMILES" -&gt; """[H][C@]12O[C@](O)([C@H](C)C[C@@H]1OC)""",
          "METABOLISM" -&gt; "The metabolism 500"
        ))
    )
  val df  = spark.createDataFrame(spark.sparkContext.parallelize(rows), recordSchema)
  </code></pre>
  
  <p>Working with that in Batch more works as a charm, no issue. </p>
  
  <p>Now I'm try to move in streaming mode using MemoryStream for testing. I added the following: </p>
  
  <blockquote>
  <pre><code>implicit val ctx = spark.sqlContext
  val intsInput = MemoryStream[Row]
  </code></pre>
  </blockquote>
  
  <p>But the compiler complain with the as follows: </p>
  
  <blockquote>
    <p><strong>No implicits found for parameter evidence$1: Encoder[Row]</strong></p>
  </blockquote>
  
  <p>Hence, my question: What should I do here to get that working</p>
  
  <p>Also i saw that if I add the following import the error goes away: </p>
  
  <blockquote>
    <p>import spark.implicits._</p>
  </blockquote>
  
  <p>Actually, I now get the following warning instead of an error</p>
  
  <blockquote>
    <p>Ambiguous implicits for parameter evidence$1: Encoder[Row]</p>
  </blockquote>
  
  <p>I do not understand the encoder mechanism well and would appreciate if someone could explain to me how not to use those implicits. The reason being that I red the following in a book when it comes to the creation of DataFrame from Rows.</p>
  
  <p>Recommended appraoch:</p>
  
  <pre><code>val myManualSchema = new StructType(Array(
    new StructField("some", StringType, true),
    new StructField("col", StringType, true),
    new StructField("names", LongType, false)))
  val myRows = Seq(Row("Hello", null, 1L))
  val myRDD = spark.sparkContext.parallelize(myRows)
  val myDf = spark.createDataFrame(myRDD, myManualSchema)
  myDf.show()
  </code></pre>
  
  <p>And then the author goes on with this: </p>
  
  <blockquote>
    <p>In Scala, we can also take advantage of Spark’s implicits in the
    console (and if you import them in your JAR code) by running toDF on a
    Seq type. This does not play well with null types, so it’s not
    necessarily recommended for production use cases.</p>
  </blockquote>
  
  <pre><code>val myDF = Seq(("Hello", 2, 1L)).toDF("col1", "col2", "col3")
  </code></pre>
  
  <p>If someone could take the time to explain what is happening in my scenario when i use the implicit, and if it is rather safe to do so, or else is there a way to do it more explicitly without importing the implicit. </p>
  
  <p>Finally, if someone could point me to a good doc around Encoder and Spark Type mapping that would be great.</p>
  
  <p><strong>EDIT1</strong></p>
  
  <p>I finally got it to work with</p>
  
  <pre><code>  implicit val ctx = spark.sqlContext
    import spark.implicits._
    val rows = MemoryStream[Map[String,String]]
    val df = rows.toDF()
  </code></pre>
  
  <p>Although my problem here is that i am not confident about what I am doing. It seems to me that it is like in some situation I need to create a DataSet to be able to convert it in an DF[ROW] with toDF conversion. I understood that working with DS is typeSafe but slower than with DF. So why this intermediary with DataSet? This is not the first time that i see that in Spark Structured Streaming. Again if someone could help me with those, that would be great.</p>