首页 > 解决方案 > 管道分离柱上的Py火花连接

问题描述

我有两个要加入的数据框。问题是其中一个表具有管道分隔的字符串,其中一个值是我想要加入的值。我如何在 Pyspark 中使用它。以下是表 A 的示例

+-------+--------------------+
|id     |      name          |
+-------+--------------------+
| 613760|123|test|test2      |
| 613740|456|ABC             |
| 598946|OMG|567             | 

表 B 有

+-------+--------------------+
|join_id|           prod_type|                           
+-------+--------------------+
| 123   |Direct De           |
| 456   |Direct              |
| 567   |In                  | 

预期结果 - 当表 A 的管道分隔 ID 与表 B 的值匹配时,连接表 A 和表 B。例如 TableA.id - 613760 名称有 123|test,我想加入表 B 的连接 ID 123,同样是 456 和 567。

结果表

+--------------------+-------+
|      name          |join_Id|
+-------+------------+-------+
|123|test|test2      |123    |
|456|ABC             |456    |
|OMG|567             |567    |

有人可以帮我解决这个问题。我对 pyspark 比较陌生,我正在学习

标签: pysparkpyspark-sqlpyspark-dataframes

解决方案


要解决您的问题,您需要:

请看下面的代码:

import pyspark.sql.functions as f

#First create the dataframes to test solution
table_A = spark.createDataFrame([(613760, '123|test|test2' ), (613740, '456|ABC'), (598946, 'OMG|567' )], ["id", "name"])
# +-------+--------------------+
# |id     |      name          |
# +-------+--------------------+
# | 613760|123|test|test2      |
# | 613740|456|ABC             |
# | 598946|OMG|567             | 

table_B = spark.createDataFrame([('123', 'Direct De' ), ('456', 'Direct'), ('567', 'In' )], ["join_id", "prod_type"])
# +-------+--------------------+
# |join_id|           prod_type|                           
# +-------+--------------------+
# | 123   |Direct De           |
# | 456   |Direct              |
# | 567   |In                  | 

result = table_A \
    .select(
        'name',
        f.posexplode(f.split(f.col('name'),'\|')).alias('pos', 'join_id')) \
    .join(table_B, on='join_id', how='inner') \
    .select('name', 'join_id')

result.show(10, False)
# +--------------+-------+
# |name          |join_id|
# +--------------+-------+
# |123|test|test2|123    |
# |456|ABC       |456    |
# |OMG|567       |567    |
# +--------------+-------+

希望有效。随着您在 Pyspark 中不断进步。我会建议您完成其中的功能,pyspark.sql.functions这将使您的技能更上一层楼。


推荐阅读