Ytukyg

Spark version 1.60, Scala version 2.10.5.

I have a spark-sql dataframe df like this,

+-------------------------------------------------+

|addess         | attributes                      | 

+-------------------------------------------------+

|1314 44 Avenue |   Tours, Mechanics, Shopping    |                                                                                                                                 

|115 25th Ave   |   Restaurant, Mechanics, Brewery|                                                                 

+-------------------------------------------------+

From this dataframe, I would like values as below,

Tours, Mechanics, Shopping, Brewery

If I do this,

df.select(df("attributes")).collect().foreach(println)

I get,

[Tours, Mechanics, Shopping]

[Restaurant, Mechanics, Brewery]

I thought I could use flatMapinstead found this, so, tried to put this into a variable using,

val allValues = df.withColumn(df("attributes"), explode("attributes"))

but I am getting an error:

error: type mismatch;

found:org.apache.spark.sql.column

required:string

I was thinking if I can get an output using explode I can use distinct to get the unique values after flattening them.

How can I get the desired output?

I strongly recommend you to use spark 2.x version. In Cloudera, when you issue "spark-shell", it launches 1.6.x version.. however, if you issue "spark2-shell", you get the 2.x shell. Check with your admin

But if you need with Spark 1.6 and rdd solution, try this.

import spark.implicits._

import scala.collection.mutable._

val df = Seq(("1314 44 Avenue",Array("Tours", "Mechanics", "Shopping")),

              ("115 25th Ave",Array("Restaurant", "Mechanics", "Brewery"))).toDF("address","attributes")

df.rdd.flatMap( x => x.getAs[mutable.WrappedArray[String]]("attributes") ).distinct().collect.foreach(println)

Results:

Brewery

Shopping

Mechanics

Restaurant

Tours

If the "attribute" column is not an array, but comma separated string, then use the below one which gives you same results

val df = Seq(("1314 44 Avenue","Tours,Mechanics,Shopping"),

  ("115 25th Ave","Restaurant,Mechanics,Brewery")).toDF("address","attributes")

df.rdd.flatMap( x => x.getAs[String]("attributes").split(",") ).distinct().collect.foreach(println)

The problem is that withColumn expects a String in its first argument (which is the name of the added column), but you're passing it a Column here df.withColumn(df("attributes").

You only need to pass "attributes" as a String.

Additionally, you need to pass a Column to the explode function, but you're passing a String - to make it a column you can use df("columName") or the Scala shorthand $ syntax, $"columnName".

Hope this example can help you.

import org.apache.spark.sql.functions._

val allValues = df.select(explode($"attributes").as("attributes")).distinct

Note that this will only preserve the attributes Column, since you want the distinct elements on that one.