在PySpark中使用JOIN和多个WHERE条件更新列值

ioekq8ef 于 2022-11-21 发布在 Spark

关注(0)|答案(1)|浏览(188)

我有一个SQL查询，我试图将其转换为PySpark，其中包含一些join和多个where条件：

UPDATE COMPANY1
INNER JOIN COMPANY2
ON COMPANY1.C1_PROFIT = COMPANY2.C2_PROFIT 
SET COMPANY2.C2_TARGET = "1"
WHERE (((COMPANY2.C2_TARGET) Is Null)
  AND ((COMPANY1.C1_SALES) Is Null)
  AND ((COMPANY2.C2_PROFIT) Is Not Null));

我正在尝试执行的PySpark查询（df_1-〉COMPANY2 & df_2-〉Company1）：

join = ((df_1.C2_PROFIT == df_2.C1_PROFIT) & \
  (df_1.C2_TARGET=='') & \
  (df_2.C1_SALES=='') & \
  (df_1.C2_PROFIT!=''))
df_1 = (df_1.alias('a')
  .join(df_2.alias('b'), join, 'left')
  .select(
    *[c for c in df_1.columns if c != 'C2_TARGET'],
    F.expr("nvl2(b.C1_PROFIT, '1', a.C2_TARGET) C2_TARGET")
  )
)

但是我仍然在列“C2_TARGET”中得到null值。
有关信息：列“C1_Profit”是无空值的，但在“C2_Profit”中，我们有时会有null以及值。
输入示例：

+------------------+--------------+
|  C1_PROFIT       |C1_SALES      |
+------------------+--------------+
|5637              |     Positive |
|7464              |              |
|43645             |              |
|64657             |      Growth P|
+------------------+--------------+

+------------------+--------------+
|  C2_PROFIT       |C2_TARGET     |
+------------------+--------------+
|                  |              |
|7464              |              |
|43645             |              |
|64657             |              |
+------------------+--------------+

预期结果：

join_on = (df_1.C1_PROFIT == df_2.C2_PROFIT) & \   --JOIN CONDITION
            (df_1.C1_REVENUE == df_3.C3_REVENUE_BREAK) & \  --JOIN CONDITION
            (df_1.C1_LOSS == df_4.C4_TOTAL_LOSS) & \        --JOIN CONDITION
            ((df_4.MARGIN_OF_COMPANY) > (df_3.LAST_YEAR_MARGIN))   --WHERE CONDITION
df = (df_1.alias('a')
    .join(df_2.alias('b'), join_on, 'left')
    .join(df_3.alias('c'), join_on, 'left')
    .join(df_4.alias('c'), join_on. 'left')
    .select(
    *[c for c in df_2.columns if c != 'C2_TARGET'],
    F.expr("nvl2(b.C2_PROFIT, '1', a.C2_TARGET) C2_TARGET")

pyspark

来源：https://stackoverflow.com/questions/74371222/update-a-column-value-using-join-and-multiple-where-conditions-in-pyspark

1条答案

按热度按时间

cetgtptt1#

在这个答案中，你有一个如何做的例子

UPDATE A INNER JOIN B
...
SET A...

中的每一个
在您的情况下，您 SET B... ：

UPDATE A INNER JOIN B
...
SET B...

格式

您已正确切换数据帧的顺序。 *

不正确的是 '' 与 null 不同。您必须在条件中使用 .isNull() 和 .isNotNull() 。
输入示例：

from pyspark.sql import functions as F
df_1 = spark.createDataFrame(
    [(5637, 'Positive'),
     (7464, None),
     (43645, None),
     (64657, 'Growth P')],
    ['C1_PROFIT', 'C1_SALES'])

df_2 = spark.createDataFrame(
    [(None, None),
     (7464, None),
     (43645, None),
     (64657, None)],
    'C2_PROFIT int, C2_TARGET string')

格式
脚本：

join_on = (df_1.C1_PROFIT == df_2.C2_PROFIT) & \
          df_2.C2_TARGET.isNull() & \
          df_1.C1_SALES.isNull() & \
          df_2.C2_PROFIT.isNotNull()
df = (df_2.alias('a')
    .join(df_1.alias('b'), join_on, 'left')
    .select(
        *[c for c in df_2.columns if c != 'C2_TARGET'],
        F.expr("nvl2(b.C1_PROFIT, '1', a.C2_TARGET) C2_TARGET")
    )
)

df.show()
# +---------+---------+
# |C2_PROFIT|C2_TARGET|
# +---------+---------+
# |     null|     null|
# |     7464|        1|
# |    64657|     null|
# |    43645|        1|
# +---------+---------+

格式

赞(0）回复(0）举报 2022-11-21

我来回答

在PySpark中使用JOIN和多个WHERE条件更新列值

1条答案

相关问题

热门标签

最新问答