Pyspark 加入管道分隔列

Question

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

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

TABLE B 有

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

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

结果Table

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

谁能帮我解决这个问题。我对 pyspark 比较陌生，正在学习

Answer 1

要解决您的问题，您需要：

split那些"pipeline separated strings"
然后将这些值分解成单独的行。 posexplode 会为你做到这一点 http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.html#pyspark.sql.functions.posexplode
从那里 "inner join" 和
最后 "select" 可以解决剩下的问题。

查看下面的代码：

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 中的功能，这将使您的技能更上一层楼。

Pyspark 加入管道分隔列

Py spark join on pipeline separated column

pyspark

pyspark-sql

pyspark-dataframes