last day (15 days later) » 

6:28 AM
1
A: org.apache.spark.SparkException: Job aborted due to stage failure - OOM Exception

mrsrinivasI would recommend you use DataFame(in Spark 2.0 i.e DataSet[Row]) as is, because DataSet uses Encoders so that it will have very little memory footprint than RDD. val destination = spark.read .options(options) .format("jdbc") .load() If you want concat columns by delimiter you can...

 
As you suggested I changed it from .rdd to .toDF after that one of my transformation error value subtractByKey is not a member of org.apache.spark.sql.Dataset[(String, String)] Code val extra_in_source = source_primary_key.subtractByKey(destination_primary_key)
toDS() is not there I have also used import org.apache.spark.sql.Dataset
 
toDS() will be available only after import spark.implicits._
 
yes I have used that ` import spark.implicits._` spark is my sparksession
 
Instead of subtract by key you use df.filter(expr). Ideally that expression should be in way to subtract records on df (generally it will be !=== expression)
 
Dataset is like (1,(1,john)) and (2,(2,steve)) so on so I used SubtractBykey
 
6:28 AM
Try create a dataframe or dataset rather than rdd(pair rdd)
 
I have changed it to .toDF() after that only its throwing error on SubtractByKey
 
SubtractByKey is not available for DataFrame
 
Hi Srinivas thanks for the reply and help
 
For DataFrame or SQL you can use something like this

`SELECT s.id, s.name FROM source_primary_key s LEFT JOIN destination_primary_key d ON s.id = d.id WHERE s.id IS NULL`
 
but its having 151 columns I just gave the example
and in most cases we wont even know how many columns
 
6:38 AM
Then the query will be something like this.

`SELECT s.* FROM source_primary_key s LEFT JOIN destination_primary_key d ON s.id = d.id WHERE s.id IS NULL`
 
I wrote like val newquery="SELECT s.* FROM "+source+" s LEFT JOIN "+destination+" d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL"
but how to execute where both source and destination were Dataframes
 
yeah, apply that query spark.sql("your sql") and make sure the dataframe is registered as tables
 
Exception in thread "main" org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input '[' expecting {<EOF>, ',', 'FROM', 'WHERE', 'GROUP', 'ORDER', 'HAVING', 'LIMIT', 'LATERAL', 'WINDOW', 'UNION', 'EXCEPT', 'MINUS', 'INTERSECT', 'SORT', 'CLUSTER', 'DISTRIBUTE'}(line 1, pos 16)

== SQL ==
SELECT s.* FROM [value: string] s LEFT JOIN [value: string] d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL
 
have registered source and destination dataframes?
 
yh
val destination = spark.read.options(options).jdbc(options("url"), options("dbtable"), new java.util.Properties()).toDF().map(_.mkString(","))
 
6:51 AM
what is the name you have given?
 
val source = sparkDestination.read.options(options1).jdbc(options1("url"), options1("dbtable"), new java.util.Properties()).toDF().map(_.mkString(","))
val newquery="SELECT s.* FROM "+source+" s LEFT JOIN "+destination+" d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL"
spark.sql(newquery).take(10).foreach(println)
 
var destination = spark.read.options(options).jdbc(options("url"), options("dbtable"), new java.util.Properties()).load()

destination = destination
.withColumn("column", concat_ws(",", destination.cols: _*))
.select("column")
.persist(StorageLevel.DISK_ONLY)


destination.createOrReplaceTempView("destination")


do the same for other source dataset also

....


then

spark.sql("SELECT s.* FROM source s LEFT JOIN destination d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL")
I don't why are you using .toDF().map(_.mkString(",")) always
 
okay I'm doing
 
destination = destination
.withColumn("column", concat_ws(",", destination.cols: _*))
.select("EMPLOYEE_ID", "column") // selecting two columns
.persist(StorageLevel.DISK_ONLY)
 
value load is not a member of org.apache.spark.sql.DataFrame
 
6:57 AM
try this

val destination = spark.read
.options(options)
.format("jdbc")
.load()
var destination = spark.read
.options(options)
.format("jdbc")
.load()
 
var destination = spark.read.options(options).jdbc(options("url"), options("dbtable"), new java.util.Properties()).format("jdbc").load()
value format is not a member of org.apache.spark.sql.DataFrame
 
var destination = spark.read.options(options).jdbc(options("url"), options("dbtable"), new java.util.Properties())

should be fine
Let's not mix the all code
 
not found: value concat_ws
nope I didnt
var destination = spark.read.options(options).jdbc(options("url"), options("dbtable"), new java.util.Properties())

destination = destination
.withColumn("column", concat_ws(",", destination.cols: _*))
.select("EMPLOYEE_ID", "column")
.persist(StorageLevel.DISK_ONLY)
 
import org.apache.spark.sql.functions._
 
now its throwing error in second line
 
7:02 AM
import org.apache.spark.sql.functions._

should works
 
value cols is not a member of org.apache.spark.sql.DataFrame
in destination.cols:
 
Can you try to resolve these compilation errors as much as you can
 
yes I'm trying
 
it would be great to know how to resolve them
This should works

.withColumn("column", concat_ws(", ", destination.columns.map(destination.col(_)).toSeq : _*))
 
Running it
Parsing command: SELECT s.* FROM [EMPLOYEE_ID: string, Name: string ... 10 more fields] s LEFT JOIN [EMPLOYEE_ID: string, Name: string ... 10 more fields] d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL
Exception in thread "main" org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input '[' expecting {<EOF>, ',', 'FROM', 'WHERE', 'GROUP', 'ORDER', 'HAVING', 'LIMIT', 'LATERAL', 'WINDOW', 'UNION', 'EXCEPT', 'MINUS', 'INTERSECT', 'SORT', 'CLUSTER', 'DISTRIBUTE'}(line 1, pos 16)
now in join query its not taking the source and destination dataframes
 
7:12 AM
Have you did


destination.createOrReplaceTempView("destination_table")
 
yeah
 
spark.sql("SELECT s.* FROM source_table s LEFT JOIN destination_table d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL")
 
sorry mybad
 
I suspect you are concatenating df with SQL. use the registered table
 
I added it like +"source"+
can't I use take()
how to check the output
 
7:16 AM
you can use `.show()`

spark.sql("SELECT s.* FROM source_table s LEFT JOIN destination_table d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL").show()

OR

var result=spark.sql("SELECT s.* FROM source_table s LEFT JOIN destination_table d ON s.EMPLOYEE_ID = d.EMPLOYEE_ID WHERE s.EMPLOYEE_ID IS NULL")

result.show()
 
bro only this I can see
+-----------+----+----------+---------+--------+----------+------+---+----------+------+-------+------+
|EMPLOYEE_ID|Name|first_name|last_name|vertical|experience|gender|DOB|percentage|salary|Pincode|column|
+-----------+----+----------+---------+--------+----------+------+---+----------+------+-------+------+
+-----------+----+----------+---------+--------+----------+------+---+----------+------+-------+------+
 
please add sample dataset for source and destination

expected result as well
 
ok
1,Ravi kumar,Ravi ,kumar,MSO ,1,M,17-01-1994,74.5,24000.78,Alabama
2,Shekhar shudhanshu,Shekhar,shudhanshu,Manulife ,2,M,18-01-1994,76.34,250000,Alaska
3,Preethi Narasingam,Preethi,Narasingam,Retail ,3,F,19-01-1994,77.45,270000.01,Arizona
source
1,Ravi kumar,Revi ,kumar,MSO ,1,M,17-01-1994,74.5,24000.78,Alabama
1,Ravi1 kumar,Revi ,kumar,MSO ,1,M,17-01-1994,74.5,24000.78,Alabama
1,Ravi2 kumar,Revi ,kumar,MSO ,1,M,17-01-1994,74.5,24000.78,Alabama
2,Shekhar shudhanshu,Shekhar,shudhanshu,Manulife ,2,M,18-01-1994,76.34,250000,Alaska
3,Preethi Narasingam1,Preethi,Narasingam,Retail ,3,F,19-01-1994,77.45,270000.01,Arizona
destination
source Destination
1 1
2 2
3 3
4
so I need 4 from source
that is extra in source
bro you ther
there*
 

last day (15 days later) »