Spark: fetch data from complex dataframe schema with map
<p>I've got a following structure</p>

<pre><code>json.select($"comments").printSchema

 root
 |-- comments: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- comment: struct (nullable = true)
 |    |    |    |-- date: string (nullable = true)
 |    |    |    |-- score: string (nullable = true)
 |    |    |    |-- shouts: array (nullable = true)
 |    |    |    |    |-- element: string (containsNull = true)
 |    |    |    |-- tags: array (nullable = true)
 |    |    |    |    |-- element: string (containsNull = true)
 |    |    |    |-- text: string (nullable = true)
 |    |    |    |-- username: string (nullable = true)
 |    |    |-- subcomments: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- date: string (nullable = true)
 |    |    |    |    |-- score: string (nullable = true)
 |    |    |    |    |-- shouts: array (nullable = true)
 |    |    |    |    |    |-- element: string (containsNull = true)
 |    |    |    |    |-- tags: array (nullable = true)
 |    |    |    |    |    |-- element: string (containsNull = true)
 |    |    |    |    |-- text: string (nullable = true)
 |    |    |    |    |-- username: string (nullable = true)
</code></pre>

<p>I would like to get an array/list [username, score, text] of comment. Normally, in pyspark I would do something like this</p>

<pre><code>comments = json
 .select("comments")
 .flatMap(lambda element: 
    map(lambda comment: 
      Row(username = comment.username, 
          score = comment.score, 
          text = comment.text), 
      element[0])
 .toDF()
</code></pre>

<p>But, when I try the same approach in scala</p>

<pre><code>json.select($"comments").rdd.map{row: Row =&gt; row(0)}.take(3)
</code></pre>

<p>I have some weird output</p>

<pre><code>Array[Any] =
Array(
  WrappedArray([[stirng,string,WrappedArray(),WrappedArray(),,string] ...],  ...)
</code></pre>

<p>Is there any way to perform that task in scala as easy as it's done with python?</p>

<p>Also, how to iterate WrappedArray like an Array/List, I'm having an error like this</p>

<pre><code>rror: scala.collection.mutable.WrappedArray.type does not take parameters
</code></pre>

<p>How about using statically typed <code>Dataset</code> instead?</p>

<pre><code>case class Comment(
    date: String, score: String,
    shouts: Seq[String], tags: Seq[String],
    text: String, username: String
)

df
  .select(explode($"comments.comment").alias("comment"))
  .select("comment.*")
  .as[Comment]
  .map(c =&gt; (c.username, c.score, c.date))
</code></pre>

<p>which can be further simplified if you don't depend on REPL:</p>

<pre><code>df
  .select("comments.comment")
  .as[Seq[Comment]]
  .flatMap(_.map(c =&gt; (c.username, c.score, c.text)))
</code></pre>

<p>If you really want to deal with <code>Rows</code> use typed getters:</p>

<pre><code>df.rdd.flatMap(
  _.getAs[SR]("comments")
    .map(_.getAs[Row]("comment"))
    .map {
      // You could also _.getAs[String]("score") or getString(0)
      case Row(_, score: String, _, _, text: String, username: String) =&gt; 
        (username, score, text)
    }
)
</code></pre>

python scala apache-spark pyspark