Einschließlich der null-Werte in einen Apache-Spark-Beitreten

Möchte ich um null-Werte in einen Apache Spark beitreten. Funken sind nicht enthalten Zeilen mit null-standardmäßig.

Hier ist die Standard-Funke-Verhalten.

val numbersDf = Seq(
  ("123"),
  ("456"),
  (null),
  ("")
).toDF("numbers")

val lettersDf = Seq(
  ("123", "abc"),
  ("456", "def"),
  (null, "zzz"),
  ("", "hhh")
).toDF("numbers", "letters")

val joinedDf = numbersDf.join(lettersDf, Seq("numbers"))

Hier ist die Ausgabe von joinedDf.show():

+-------+-------+
|numbers|letters|
+-------+-------+
|    123|    abc|
|    456|    def|
|       |    hhh|
+-------+-------+

Dies ist die Ausgabe, die ich möchte:

+-------+-------+
|numbers|letters|
+-------+-------+
|    123|    abc|
|    456|    def|
|       |    hhh|
|   null|    zzz|
+-------+-------+
InformationsquelleAutor Powers | 2017-01-18
  1. 52

    Spark bietet eine Besondere NULL sicher Gleichheits-operator: 

    numbersDf
  .join(lettersDf, numbersDf("numbers") <=> lettersDf("numbers"))
  .drop(lettersDf("numbers"))
    +-------+-------+
|numbers|letters|
+-------+-------+
|    123|    abc|
|    456|    def|
|   null|    zzz|
|       |    hhh|
+-------+-------+

    Vorsichtig sein, es nicht zu verwenden mit Spark 1.5 oder früher. Vor Spark 1.6 ist es erforderlich, ein Kartesisches Produkt (SPARK-11111 - Fast-null-safe join).

    In Funke 2.3.0 oder später können Sie Column.eqNullSafe im PySpark:

    numbers_df = sc.parallelize([
    ("123", ), ("456", ), (None, ), ("", )
]).toDF(["numbers"])

letters_df = sc.parallelize([
    ("123", "abc"), ("456", "def"), (None, "zzz"), ("", "hhh")
]).toDF(["numbers", "letters"])

numbers_df.join(letters_df, numbers_df.numbers.eqNullSafe(letters_df.numbers))
    +-------+-------+-------+
|numbers|numbers|letters|
+-------+-------+-------+
|    456|    456|    def|
|   null|   null|    zzz|
|       |       |    hhh|
|    123|    123|    abc|
+-------+-------+-------+

    sowie %<=>% im SparkR:

    numbers_df <- createDataFrame(data.frame(numbers = c("123", "456", NA, "")))
letters_df <- createDataFrame(data.frame(
  numbers = c("123", "456", NA, ""),
  letters = c("abc", "def", "zzz", "hhh")
))

head(join(numbers_df, letters_df, numbers_df$numbers %<=>% letters_df$numbers))
      numbers numbers letters
1     456     456     def
2    <NA>    <NA>     zzz
3                     hhh
4     123     123     abc

    Mit SQL (Funke 2.2.0+) können Sie IS NOT DISTINCT FROM:

    SELECT * FROM numbers JOIN letters 
ON numbers.numbers IS NOT DISTINCT FROM letters.numbers

    Dies kann verwendet werden, mit DataFrame - API:

    numbersDf.alias("numbers")
  .join(lettersDf.alias("letters"))
  .where("numbers.numbers IS NOT DISTINCT FROM letters.numbers")
    • Danke. Dies ist eine weitere gute Antwort, verwendet die <=> Betreiber. Wenn Sie ein mehrere Spalte join-Bedingungen können verkettet werden mit der && Betreiber.
    • Meiner Erfahrung nach (Funke 2.2.1 auf Amazon-Kleber), die SQL-syntax ist die gleiche wie die Scala: SELECT * FROM Rufnummern JOIN Buchstaben AUF zahlen.zahlen <=> Briefe.zahlen
    InformationsquelleAutor user6910411
  2. 8
    val numbers2 = numbersDf.withColumnRenamed("numbers","num1") //rename columns so that we can disambiguate them in the join
val letters2 = lettersDf.withColumnRenamed("numbers","num2")
val joinedDf = numbers2.join(letters2, $"num1" === $"num2" || ($"num1".isNull &&  $"num2".isNull) ,"outer")
joinedDf.select("num1","letters").withColumnRenamed("num1","numbers").show  //rename the columns back to the original names
    InformationsquelleAutor jasonS
  3. 0

    Versuchen Sie die folgende Methode, um die null-Zeilen, um das Ergebnis der JOIN-operator:

    def nullSafeJoin(leftDF: DataFrame, rightDF: DataFrame, columns: Seq[String], joinType: String): DataFrame = {

    var columnsExpr: Column = leftDF(columns.head) <=> rightDF(columns.head)

    columns.drop(1).foreach(column => {
        columnsExpr = columnsExpr && (leftDF(column) <=> rightDF(column))
    })

    var joinedDF: DataFrame = leftDF.join(rightDF, columnsExpr, joinType)

    columns.foreach(column => {
        joinedDF = joinedDF.drop(leftDF(column))
    })

    joinedDF
}
    InformationsquelleAutor K L
Schreibe einen Kommentar