Created 02-06-2017 02:17 AM
I have been trying to make the following Dataframe query work but its not giving me the results. Can someone please help?
Is there a reference guide to see all the syntax to create dataframe queries? Here this is what I want - My dataframe df has many cols among which 4 are -
For all rows, if deviceFlag is set to 1, then form a string "device#domain" and store that value in a new col "id". Else, just take the value in the "device" col and store it in the new "id" col without any transformation. So I am creating a new column with the .withColumn and setting value for each row depending upon the corresponding values of "device" and "deviceFlag" cols in that row.
DataFrame result= df.withColumn("id",
((Column) when(df.col("deviceFlag").$eq$eq$eq(1),
concat(df.col("device"), lit("#"),
df.col("domain")))).otherwise(df.col("device")).alias("id"));
Created 02-07-2017 02:38 PM
I think your solution is ok. Probably you have problem with the input data (Try to call df.show() and df.printSchema() with the original df before the transformation). I tried out your code with this class:
package com.hortonworks.spark.example; import static org.apache.spark.sql.functions.when; import static org.apache.spark.sql.functions.concat; import static org.apache.spark.sql.functions.lit; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.SparkConf; import org.apache.spark.sql.Column; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SQLContext; import java.io.Serializable; import java.util.Arrays; import java.util.List; public class SparkStarter { public static void main(String[] args) { new SparkStarter().run(); } private void run() { SparkConf conf = new SparkConf().setAppName("test").setMaster("local[*]"); JavaSparkContext sc = new JavaSparkContext(conf); List<Record> data = Arrays.asList(new Record(1, null, "b"),new Record(1, "a", "b"), new Record(0, "b", "c")); JavaRDD<Record> distData = sc.parallelize(data); SQLContext sql = new SQLContext(sc); Dataset<Row> df = sql.createDataFrame(distData, Record.class); df.withColumn("id", ((Column) when(df.col("deviceFlag").$eq$eq$eq(1), concat(df.col("device"), lit("#"), df.col("domain")))).otherwise(df.col("device")).alias("id")).show(); } public static class Record implements Serializable { int deviceFlag; String device; String domain; public int getDeviceFlag() { return deviceFlag; } public void setDeviceFlag(int deviceFlag) { this.deviceFlag = deviceFlag; } public String getDevice() { return device; } public void setDevice(String device) { this.device = device; } public String getDomain() { return domain; } public void setDomain(String domain) { this.domain = domain; } public Record(int deviceFlag, String device, String domain) { this.deviceFlag = deviceFlag; this.device = device; this.domain = domain; } } }
And it worked well:
+------+----------+------+----+ |device|deviceFlag|domain| id| +------+----------+------+----+ | null| 1| b|null| | a| 1| b| a#b| | b| 0| c| b| +------+----------+------+----+
What was your problem exactly?
Created 02-07-2017 09:51 PM
I am getting a NPE as below
df is a dataframe that I have from previous joins with other dataframes DataFrame df = df1.join(....) // On doing tis I get the following NPE df.withColumn("id", ((Column) when(df.col("deviceFlag").$eq$eq$eq(1), concat(df.col("device"), lit("#"), df.col("domain")))).otherwise(df.col("device")).alias("id")).show(); The line 235 in ApplDriver is df.col("domain")))).otherwise(df.col("device")).alias("id")).show(); I have checked some input values of domain and device and they are coming fine. NPE tells me its expecting some object? But cols are there, "id" is anyway a new col so its not defined earlier and df is ofcourse the resulting DF from prev steps..Then what can cause the NPE? Its not going forward because of the NPE. 17/02/07 15:39:18 ERROR ApplicationMaster: User class threw exception: java.lang.NullPointerException java.lang.NullPointerException at com.Driver.ApplDriver.lambda$main$1acd672$1(ApplDriver.java:235) at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3.apply(JavaDStreamLike.scala:335) at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3.apply(JavaDStreamLike.scala:335) at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:661) at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:661) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:50) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:50) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:50) at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:426) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:49) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:49) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:49) at scala.util.Try$.apply(Try.scala:161) at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39) at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:227) at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:227) at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:227) at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57) at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:226) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745)
Created 02-07-2017 10:55 PM
Some more testing showed that if I just add the new col with concat, it concats the values and shows up fine. But on using when() and otherwise() it is throwing an NPE
Created 02-09-2017 07:08 PM
Thanks this worked. Somehow a when() function was declared in my class returning null and hence the NPE.