Title: Scala - How to convert a Dataset[Row] to a column that can be added to a Dataframe
Tags: scala;apache-spark;dataframe;dataset
Question: I'm trying to add a dataframe of one column to a larger dataframe however the issue with the first dataframe is that after creating it and trying to add it to the main dataframe via the command:
df.withColumn("name", dataframe)
I get the error:
**found   : org.apache.spark.sql.DataFrame
(which expands to)  org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
required: org.apache.spark.sql.Column**
I understand that a Dataset[Row] is supposed to be synonymous with a Dataframe however I'm not sure how to get around this error.
For context, a (really) watered down version of my code is below:
// test function - will be used as part of the main script below
def Test(inputone: Double, inputtwo: Double): Double = {
var test = (2 * inputone) + inputtwo
test
For the main script (i.e. where the problem lies)
//Importing the data via CSV
var df = sqlContext.read.format("csv").option("header",     "true").option("inferSchema", "true").load("/root/file.csv")
To give a context of what the data looks like:
df: org.apache.spark.sql.DataFrame = [ID: int, blue: int ... 8 more fields]
+---+----+------+-----+------+------+----+---+-----+-----+
| ID|blue|purple|green|yellow|orange|pink|red|white|black|
+---+----+------+-----+------+------+----+---+-----+-----+
|  1| 500|    44|    0|     0|     3|   0|  5|   43|    2|
|  2| 560|    33|    1|     0|     4|   0| 22|   33|    4|
|  3| 744|    44|    1|    99|     3|1000| 78|   90|    0|
+---+----+------+-----+------+------+----+---+-----+-----+
|-- ID: integer (nullable = true)
|-- blue: integer (nullable = true)
|-- purple: integer (nullable = true)
|-- green: integer (nullable = true)
|-- yellow: integer (nullable = true)
|-- orange: integer (nullable = true)
|-- pink: integer (nullable = true)
|-- red: integer (nullable = true)
|-- white: integer (nullable = true)
|-- black: integer (nullable = true)
From then on, the script continues
// Creating a list for which columns to draw from the main dataframe
val a = List("green", "blue")
// Creating the mini dataframe to perform the function upon
val test_df = df.select(a.map(col): _*)
// The new dataframe will now go through the 'Test' function defined above
val df_function = test_df.rdd.map(col =&gt; Test(col(0).toString.toDouble, col(1).toString.toDouble))
// Converting the RDD output back to a dataframe (of one column)
val df_convert = df_function.toDF
As a reference, the output looks as follows
+-----+
|value|
+-----+
|500.0|
|562.0|
|746.0|
+-----+
The last line of the script is to add it to the main dataframe as follows
df = df.withColumn("new column", df_convert)
But as stated above, I receive the following error:
found   : org.apache.spark.sql.DataFrame
(which expands to)  org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
required: org.apache.spark.sql.Column
//////////EDIT////////////
@user9819212 solution works for simplistic methods but when calling one a bit more complex, I get the following error
test2_udf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function5&gt;,DoubleType,Some(List(DoubleType, IntegerType, StringType, DoubleType, DoubleType)))
java.lang.ClassCastException: $anonfun$1 cannot be cast to scala.Function1
So I tried to create another simplistic version of my code with a few extra changes to the test function that is called
// test function - will be used as part of the main script below
def Test (valueone: Double, valuetwo: Integer): Double = {
val test = if(valuetwo &gt; 2000) valueone + 4000 else valueone
val fakeList = List(3000,4000,500000000)
val index = fakeList.indexWhere(x =&gt; x&gt;=valueone)
val test2 = fakeList(index - 1) * valueone
test2
val test_udf = udf(Test _)
df = df.withColumn(
"new column",
test_udf(col("green").cast("double"), col("blue").cast("integer"))
At first that seems to work but when I try to view the dataframe with the command
df.show
I get the following error
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 153.0 failed 1 times, most recent failure: Lost task 0.0 in stage 153.0 (TID 192, localhost, executor driver):
org.apache.spark.SparkException: Failed to execute user defined function($anonfun$1: (double, int) =&gt; double)
Comment: Looking at your edited Test function, there is only one mistake i.e. when the index is 0 (when `fakeList.indexWhere(x => x>=valueone)`) returns 0. You have to handle that. Otherwise the udf function looks perfect. But you can refactor the code using when inbuilt function

Comment: comments are not good for codes. can you update the question the the error and `test2_udf` function definition and how you are invoking it?

Comment: meanwhile you can check my detailed answer below :)

Comment: @Andrey Tyukin I've changed the title, is it more appropriate now or should it not reference the Dataset[Row] type?

Comment: @Ramesh Maharjan ah I may have simplified it too much again, my bad - I'm really just stuck with the following error ==>  test2_udf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(,DoubleType,Some(List(DoubleType, IntegerType, StringType, DoubleType, DoubleType)))
java.lang.ClassCastException: $anonfun$1 cannot be cast to scala.Function1 - when I put it to my main script, I can provide more details if needed

Comment: @Ramesh Maharjan upon first glance I believe you may be right, I'm without my normal computer at the moment so I cannot access my code but I will review my code with what you've mentioned in your comment when I get my computer back and I'll accept the answer if it works

Comment: The title is fine, thank you for updating it.

Here is the accepted answer: You cannot add columns from another DataFrame (or DataFrame) this way. Just use UserDefinedFunction:
import org.apache.spark.sql.functions.udf._
val test_udf = udf(Test _)
df.withColumn(
"new column",
test_udf(col("green").cast("double"), col("blue").cast("double"))
or with such simple function:
df.withColumn(
"new column",
2 * col("green").cast("double") + col("blue").cast("double")
Comment for this answer: @user9819212 you're solution works for the simplistic function I've provided but when I try to implement it to my real script, I get errors (this is my fault because I tried to simplify the script too much so it would be easier to post on StackExchange)
I'll edit my original post to explain what I mean in a minute

Here is another answer: If you go the api document it is clearly mentioned as
public DataFrame withColumn(java.lang.String colName, Column col)
Returns a new DataFrame by adding a column or replacing the existing column that has the same name.
As you can see that second argument should be a Column and you have been passing a DataFrame.
thats the cause of the issue.
And you are trying to add a column from df_convert to df. The two dataframes are totally different. For that case you will have to look at either join if you want to separate the dataframes
Or at spark functions to be used with withColumn api as Column.
Updated
Looking at your first log
test2_udf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function5&gt;,DoubleType,Some(List(DoubleType, IntegerType, StringType, DoubleType, DoubleType)))
suggests that you have your udf function defined as
def Test(valueone: Double, valuetwo: Integer, valuethree: String, valuefour: Double, valuefive: Double): Double = {
//calculation parts
val test2_udf = udf(Test _)
//Test: Test[](val valueone: Double,val valuetwo: Integer,val valuethree: String,val valuefour: Double,val valuefive: Double) =&gt; Double
//test2_udf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function5&gt;,DoubleType,Some(List(DoubleType, IntegerType, StringType, DoubleType, DoubleType)))
and your second log
java.lang.ClassCastException: $anonfun$1 cannot be cast to scala.Function1
suggests that you are passing only one argument in the test2_udf call as
df.withColumn("new column", test2_udf(col("green").cast("double"))).show(false)
//java.lang.ClassCastException: A$A30$A$A30$$anonfun$test2_udf$1 cannot be cast to scala.Function1
If you focus on cannot be cast to scala.Function1 part of the error message, it clearly suggests the number of columns passed to the udf functions
If you pass three arguments then you get following
df.withColumn("new column", test2_udf(col("green").cast("double"),col("green").cast("double"),col("green").cast("double"))).show(false)
//java.lang.ClassCastException: A$A31$A$A31$$anonfun$test2_udf$1 cannot be cast to scala.Function3
Comment for this answer: I am glad that it helped you. you can always upvote when you would be eligible ;)
Comment for this answer: I should have mentioned this earlier, but your updated part is the right answer to the second part of my question (i.e. when I started encountering errors) - so thanks :)

{'source': 'StackExchangeDataset', 'subset': 'stackoverflow.com', 'metadata': {'AnswerCount': 2, 'CommentCount': 7, 'ContentLicense': 'CC BY-SA 4.0', 'CreationDate': '2018:00:05/20/18', 'FavoriteCount': 0.0, 'PostTypeId': 1, 'Tags': '<scala><apache-spark><dataframe><dataset>', 'Title': 'Scala - How to convert a Dataset[Row] to a column that can be added to a Dataframe', 'ViewCount': 1853, 'id': 50435767, 'user_AboutMe': None, 'user_AccountId': 13611814.0, 'user_DisplayName': 'ddssuu', 'user_Reputation': 43.0, 'user_Views': 10.0, 'user_id': 9819067.0}}