Фильтрация условного объединения по значению строки

Question

У меня есть 2 кадра данных, которые мне нужно объединить.Есть 2 столбца, к которым мы присоединяемся, однако эти 2 столбца не являются идентификаторами и не создают уникальных значений строк в другом кадре данных.Например, два кадра данных будут выглядеть так:

Dataframe 1 
    product_no   dist   code
    040          wmn    aj
    040          wmn    lm
    040          wmn    mn

Dataframe 2
    p_no   vendor    code   product
    040    wmn       **     y
    040    wmn       *j     y
    040    wmn       mn     n

Запрос для двух столбцов будет выглядеть так:

df1.join(df2, $"product_no" === $"p_no" && $"dist" === $"vendor")

Это объединит все строки друг с другом и будет иметьдубликаты.Однако мы также хотим присоединиться к code Например, где код равен фактическому коду во втором кадре данных.Если мы не можем найти фактическое, то мы проверяем, заканчивается ли код на j, если это так, тогда мы присоединяемся на *j, если у нас нет ничего, что соответствует, мы затем соединяем на ** результатдолжно выглядеть так:

**Outcome**
        product_no   dist   code  p_no   vendor    code   product
        040          wmn    aj    040    wmn       *j     y
        040          wmn    lm    040    wmn       **     y
        040          wmn    mn    040    wmn       mn     n

Есть ли способ, которым я могу это сделать?

Answer 1

Проверьте это:

scala> val df3 = df1.alias("t1").join(df2.alias("t2"),$"product_no" === $"p_no" && $"dist" === $"vendor").withColumn("match", when($"t1.code"===$"t2.code",lit(1)).when(regexp_extract($"t1.code",".*j",0)=!=lit("") && regexp_extract($"t2.code",".*j",0)=!=lit(""), 2).when(regexp_extract($"t1.code",".*[^j]$",0)=!=lit("") &&  regexp_extract($"t2.code","[*][*]",0)=!=lit(""), 3).otherwise(lit(0))).filter('match > 0).toDF("product_no","dist","code1","p_no","vendor","code2","product","match")
df3: org.apache.spark.sql.DataFrame = [product_no: string, dist: string ... 6 more fields]

scala> val df4= df3.withColumn("match2", collect_set('code2) over(Window.partitionBy('product_no,'dist).orderBy('match)))
df4: org.apache.spark.sql.DataFrame = [product_no: string, dist: string ... 7 more fields]

scala> df4.show
+----------+----+-----+----+------+-----+-------+-----+------------+
|product_no|dist|code1|p_no|vendor|code2|product|match|      match2|
+----------+----+-----+----+------+-----+-------+-----+------------+
|       040| wmn|   mn| 040|   wmn|   mn|      n|    1|        [mn]|
|       040| wmn|   aj| 040|   wmn|   *j|      y|    2|    [*j, mn]|
|       040| wmn|   mn| 040|   wmn|   **|      y|    3|[*j, mn, **]|
|       040| wmn|   lm| 040|   wmn|   **|      y|    3|[*j, mn, **]|
+----------+----+-----+----+------+-----+-------+-----+------------+

scala> df4.selectExpr("*"," match in (1,2) or ( not array_contains(match2,code1) ) as match3 ").where('match3).show
+----------+----+-----+----+------+-----+-------+-----+------------+------+
|product_no|dist|code1|p_no|vendor|code2|product|match|      match2|match3|
+----------+----+-----+----+------+-----+-------+-----+------------+------+
|       040| wmn|   mn| 040|   wmn|   mn|      n|    1|        [mn]|  true|
|       040| wmn|   aj| 040|   wmn|   *j|      y|    2|    [*j, mn]|  true|
|       040| wmn|   lm| 040|   wmn|   **|      y|    3|[*j, mn, **]|  true|
+----------+----+-----+----+------+-----+-------+-----+------------+------+


scala>

Answer 2

Я тоже прошел через несколько похожих сценариев, где присоединение основано на условном флаге.

val decision: Boolean = false
val exprs = (if (decision != true) 
                   dataFrame1.col("id").equalTo(dataFrame2.col("id"))
             else dataFrame1.col("id").equalTo(dataFrame2.col("id")) 
                   and dataFrame1.col("code").equalTo(dataFrame2.col("code")))
            )
dataFrame1.join(dataFrame2, exprs).show

Пример вы можете увидеть здесь Вы можете использовать ту же логику здесь.

Вместо условного флага попробуйте значения строки.