31

... by checking whether a columns' value is in a seq.
Perhaps I'm not explaining it very well, I basically want this (to express it using regular SQL): DF_Column IN seq?

First I did it using a broadcast var (where I placed the seq), UDF (that did the checking) and registerTempTable.
The problem is that I didn't get to test it since I ran into a known bug that apparently only appears when using registerTempTable with ScalaIDE.

I ended up creating a new DataFrame out of seq and doing inner join with it (intersection), but I doubt that's the most performant way of accomplishing the task.

Thanks

EDIT: (in response to @YijieShen):
How to do filter based on whether elements of one DataFrame's column are in another DF's column (like SQL select * from A where login in (select username from B))?

E.g: First DF:

login      count
login1     192  
login2     146  
login3     72   

Second DF:

username
login2
login3
login4

The result:

login      count
login2     146  
login3     72   

Attempts:
EDIT-2: I think, now that the bug is fixed, these should work. END EDIT-2

ordered.select("login").filter($"login".contains(empLogins("username")))

and

ordered.select("login").filter($"login" in empLogins("username"))

which both throw Exception in thread "main" org.apache.spark.sql.AnalysisException, respectively:

resolved attribute(s) username#10 missing from login#8 in operator 
!Filter Contains(login#8, username#10);

and

resolved attribute(s) username#10 missing from login#8 in operator 
!Filter login#8 IN (username#10);
6
  • What's the size of the Seq roughly? Commented Apr 29, 2015 at 10:42
  • Small, currently 100 elements, and it should never go above 10k. Commented Apr 29, 2015 at 12:47
  • How about use the DSL of dataFrame instead of sql?
    – yjshen
    Commented Apr 30, 2015 at 4:19
  • @MarkoBonaci, which version of spark are you using? Does [spark-5281] also affect your version? or just because you are using ScalaIDE that causes your first attempt fail?
    – yjshen
    Commented May 1, 2015 at 3:33
  • @MarkoBonaci, please check if the answer i posted below works for you.
    – yjshen
    Commented May 1, 2015 at 5:07

2 Answers 2

16

My code (following the description of your first method) runs normally in Spark 1.4.0-SNAPSHOT on these two configurations:

  • Intellij IDEA's test
  • Spark Standalone cluster with 8 nodes (1 master, 7 worker)

Please check if any differences exists

val bc = sc.broadcast(Array[String]("login3", "login4"))
val x = Array(("login1", 192), ("login2", 146), ("login3", 72))
val xdf = sqlContext.createDataFrame(x).toDF("name", "cnt")

val func: (String => Boolean) = (arg: String) => bc.value.contains(arg)
val sqlfunc = udf(func)
val filtered = xdf.filter(sqlfunc(col("name")))

xdf.show()
filtered.show()

Output

name cnt
login1 192
login2 146
login3 72

name cnt
login3 72

5
  • I'm using 1.3.1. I can also run it in all environments except ScalaIDE. I've just tried your code and the same bug happens (the same code is run in the background, as when using sql expresson). In IDEA, it runs fine. So do we conclude that, given the bug, the only possible option in ScalaIDE is to do inner join? Or you have some alternative ideas? Commented May 1, 2015 at 8:14
  • @MarkoBonaci, or you could just move to IDEA and forget about the error :)
    – yjshen
    Commented May 1, 2015 at 13:43
  • :) Well, this is a part of an example in a book I'm working on. But, I was just thinking that maybe this wasn't such a bad thing after all. It could be like a real-life situation that you're forced to deal with a road block and find an alternate solution. From that standpoint, it's not even that important that the algorithm does not suck (I guess, alternatives must always suck, even just a little bit :) Would you mind if you became one of the characters in the book (like a supporting actor - a good samaritan from the SO)? BTW, it's Spark in Action for Manning. Commented May 1, 2015 at 15:15
  • @MarkoBonaci, just thinking broadcast variable & udf fits your requirements better, although workaround is a must sometimes. No, I don't mind :)
    – yjshen
    Commented May 1, 2015 at 15:26
  • @YijieShen I am getting an error while performing udf inside filter ..like found ColName required String ` val DS_in = DS_In.filter($"cid" === 2).join(DS_B,Seq("cid","aid"),"inner").filter(compareUsage($"usagetype"))` my udf val compareUsage: (String => Boolean) = (str: String) => if (str == "" || str == null) { true } else if ((str.toString().toUpperCase().contains("Acitve")) || (str.toString().toUpperCase() contains ("USAGE"))) true else false Can you tell me where its is going wrong
    – Anji
    Commented Feb 3, 2017 at 14:37
12
+50
  1. You should broadcast a Set, instead of an Array, much faster searches than linear.

  2. You can make Eclipse run your Spark application. Here's how:

As pointed out on the mailing list, spark-sql assumes its classes are loaded by the primordial classloader. That's not the case in Eclipse, were the Java and Scala library are loaded as part of the boot classpath, while the user code and its dependencies are in another one. You can easily fix that in the launch configuration dialog:

  • remove Scala Library and Scala Compiler from the "Bootstrap" entries
  • add (as external jars) scala-reflect, scala-library and scala-compiler to the user entry.

The dialog should look like this:

enter image description here

Edit: The Spark bug was fixed and this workaround is no longer necessary (since v. 1.4.0)

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.