Pyspark连接数据框中列的逗号分隔值

Question

所以我有两个要加入的数据框。渔获物是第二个表中存储有逗号分隔的值，其中一个与表A中的列匹配。我如何在Pyspark中使用它。下面是一个示例

表A具有

+-------+--------------------+
|deal_id|           deal_name|
+-------+--------------------+
| 613760|ABCDEFGHI           |
| 613740|TEST123             |
| 598946|OMG                 |   

表B有

+-------+---------------------------+--------------------+
|                            deal_id|           deal_type|                           
+-------+---------------------------+--------------------+
| 613760,613761,613762,613763       |Direct De           |
| 613740,613750,613770,613780,613790|Direct              |
| 598946                            |In                  |  

预期结果-当表A的交易ID与表B的逗号分隔值匹配时，将表A和表B连接起来。例如TableA.dealid-613760在表B的第一行中，我希望返回该行。

+-------+--------------------+---------------+
|deal_id|           deal_name|      deal_type|
+-------+--------------------+---------------+
| 613760|ABCDEFGHI           |Direct De      |     
| 613740|TEST123             |Direct         |
| 598946|OMG                 |In             |

感谢您的协助。我在pyspark中需要它。

谢谢。

Answer 1

样本数据

from pyspark.sql.types import IntegerType, LongType, StringType, StructField, StructType

tuples_a = [('613760', 'ABCDEFGHI'),
            ('613740', 'TEST123'),
            ('598946', 'OMG'),
           ]

schema_a = StructType([
         StructField('deal_id', StringType(), nullable=False),
         StructField('deal_name', StringType(), nullable=False)
        ])


tuples_b = [('613760,613761,613762,613763 ', 'Direct De'),
            ('613740,613750,613770,613780,613790', 'Direct'),
            ('598946', 'In'),
           ]

schema_b = StructType([
         StructField('deal_id', StringType(), nullable=False),
         StructField('deal_type', StringType(), nullable=False)
        ])        

df_a = spark_session.createDataFrame(data=tuples_a, schema=schema_a)
df_b = spark_session.createDataFrame(data=tuples_b, schema=schema_b) 

您需要拆分并分解列才能加入。

from pyspark.sql.functions import split, col, explode

df_b = df_b.withColumn('split', split(col('deal_id'), ','))\
           .withColumn('exploded', explode(col('split')))\
           .drop('deal_id', 'split')\
           .withColumnRenamed('exploded', 'deal_id')


df_a.join(df_b, on = 'deal_id', how = 'left_outer')\
    .show(10, False)

和预期结果

+-------+---------+---------+
|deal_id|deal_name|deal_type|
+-------+---------+---------+
|613760 |ABCDEFGHI|Direct De|
|613740 |TEST123  |Direct   |
|598946 |OMG      |In       |
+-------+---------+---------+