基于 PySpark 中另一个数据框的列值创建指标数组
Creating an indicator array based on other data frame's column values in PySpark
我有两个数据框:df1
+---+-----------------+
|id1| items1|
+---+-----------------+
| 0| [B, C, D, E]|
| 1| [E, A, C]|
| 2| [F, A, E, B]|
| 3| [E, G, A]|
| 4| [A, C, E, B, D]|
+---+-----------------+
和df2
:
+---+-----------------+
|id2| items2|
+---+-----------------+
|001| [A, C]|
|002| [D]|
|003| [E, A, B]|
|004| [B, D, C]|
|005| [F, B]|
|006| [G, E]|
+---+-----------------+
我想根据 items2
中的值创建一个指标向量(在 df1
中的新列 result_array
中)。该向量的长度应与 df2
中的行数相同(在此示例中,它应包含 6 个元素)。如果 items1
中的行包含 items2
对应行中的所有元素,则其元素值应为 1.0,否则应为值 0.0。结果应如下所示:
+---+-----------------+-------------------------+
|id1| items1| result_array|
+---+-----------------+-------------------------+
| 0| [B, C, D, E]|[0.0,1.0,0.0,1.0,0.0,0.0]|
| 1| [E, A, C]|[1.0,0.0,0.0,0.0,0.0,0.0]|
| 2| [F, A, E, B]|[0.0,0.0,1.0,0.0,1.0,0.0]|
| 3| [E, G, A]|[0.0,0.0,0.0,0.0,0.0,1.0]|
| 4| [A, C, E, B, D]|[1.0,1.0,1.0,1.0,0.0,0.0]|
+---+-----------------+-------------------------+
例如,在第 0 行中,第二个值为 1.0,因为 [D] 是 [B, C, D, E] 的子集,第四个值为 1.0,因为 [B, D, C] 是[B, C, D, E] 的子集。 df2
中的所有其他项目组都不是 [B、C、D、E] 的子集,因此它们的指标值为 0.0。
我尝试使用 collect() 创建 items2
中所有项目组的列表,然后应用 udf,但我的数据太大(超过 1000 万行)。
您可以这样进行,
import pyspark.sql.functions as F
from pyspark.sql.types import *
df1 = sql.createDataFrame([
(0,['B', 'C', 'D', 'E']),
(1,['E', 'A', 'C']),
(2,['F', 'A', 'E', 'B']),
(3,['E', 'G', 'A']),
(4,['A', 'C', 'E', 'B', 'D'])],
['id1','items1'])
df2 = sql.createDataFrame([
(001,['A', 'C']),
(002,['D']),
(003,['E', 'A', 'B']),
(004,['B', 'D', 'C']),
(005,['F', 'B']),
(006,['G', 'E'])],
['id2','items2'])
它给你数据帧,
+---+---------------+
|id1| items1|
+---+---------------+
| 0| [B, C, D, E]|
| 1| [E, A, C]|
| 2| [F, A, E, B]|
| 3| [E, G, A]|
| 4|[A, C, E, B, D]|
+---+---------------+
+---+---------+
|id2| items2|
+---+---------+
| 1| [A, C]|
| 2| [D]|
| 3|[E, A, B]|
| 4|[B, D, C]|
| 5| [F, B]|
| 6| [G, E]|
+---+---------+
现在,crossJoin
两个数据帧,它为您提供 df1
与 df2
的笛卡尔积。然后,在 'items1'
上 groupby
并应用 udf
以获得 'result_array'
.
get_array_udf = F.udf(lambda x,y:[1.0 if set(z) < set(x) else 0.0 for z in y], ArrayType(FloatType()))
df = df1.crossJoin(df2)\
.groupby(['id1', 'items1']).agg(F.collect_list('items2').alias('items2'))\
.withColumn('result_array', get_array_udf('items1', 'items2')).drop('items2')
df.show()
这为您提供了输出,
+---+---------------+------------------------------+
|id1|items1 |result_array |
+---+---------------+------------------------------+
|1 |[E, A, C] |[1.0, 0.0, 0.0, 0.0, 0.0, 0.0]|
|0 |[B, C, D, E] |[0.0, 1.0, 0.0, 1.0, 0.0, 0.0]|
|4 |[A, C, E, B, D]|[1.0, 1.0, 1.0, 1.0, 0.0, 0.0]|
|3 |[E, G, A] |[0.0, 0.0, 0.0, 0.0, 0.0, 1.0]|
|2 |[F, A, E, B] |[0.0, 0.0, 1.0, 0.0, 1.0, 0.0]|
+---+---------------+------------------------------+
我有两个数据框:df1
+---+-----------------+
|id1| items1|
+---+-----------------+
| 0| [B, C, D, E]|
| 1| [E, A, C]|
| 2| [F, A, E, B]|
| 3| [E, G, A]|
| 4| [A, C, E, B, D]|
+---+-----------------+
和df2
:
+---+-----------------+
|id2| items2|
+---+-----------------+
|001| [A, C]|
|002| [D]|
|003| [E, A, B]|
|004| [B, D, C]|
|005| [F, B]|
|006| [G, E]|
+---+-----------------+
我想根据 items2
中的值创建一个指标向量(在 df1
中的新列 result_array
中)。该向量的长度应与 df2
中的行数相同(在此示例中,它应包含 6 个元素)。如果 items1
中的行包含 items2
对应行中的所有元素,则其元素值应为 1.0,否则应为值 0.0。结果应如下所示:
+---+-----------------+-------------------------+
|id1| items1| result_array|
+---+-----------------+-------------------------+
| 0| [B, C, D, E]|[0.0,1.0,0.0,1.0,0.0,0.0]|
| 1| [E, A, C]|[1.0,0.0,0.0,0.0,0.0,0.0]|
| 2| [F, A, E, B]|[0.0,0.0,1.0,0.0,1.0,0.0]|
| 3| [E, G, A]|[0.0,0.0,0.0,0.0,0.0,1.0]|
| 4| [A, C, E, B, D]|[1.0,1.0,1.0,1.0,0.0,0.0]|
+---+-----------------+-------------------------+
例如,在第 0 行中,第二个值为 1.0,因为 [D] 是 [B, C, D, E] 的子集,第四个值为 1.0,因为 [B, D, C] 是[B, C, D, E] 的子集。 df2
中的所有其他项目组都不是 [B、C、D、E] 的子集,因此它们的指标值为 0.0。
我尝试使用 collect() 创建 items2
中所有项目组的列表,然后应用 udf,但我的数据太大(超过 1000 万行)。
您可以这样进行,
import pyspark.sql.functions as F
from pyspark.sql.types import *
df1 = sql.createDataFrame([
(0,['B', 'C', 'D', 'E']),
(1,['E', 'A', 'C']),
(2,['F', 'A', 'E', 'B']),
(3,['E', 'G', 'A']),
(4,['A', 'C', 'E', 'B', 'D'])],
['id1','items1'])
df2 = sql.createDataFrame([
(001,['A', 'C']),
(002,['D']),
(003,['E', 'A', 'B']),
(004,['B', 'D', 'C']),
(005,['F', 'B']),
(006,['G', 'E'])],
['id2','items2'])
它给你数据帧,
+---+---------------+
|id1| items1|
+---+---------------+
| 0| [B, C, D, E]|
| 1| [E, A, C]|
| 2| [F, A, E, B]|
| 3| [E, G, A]|
| 4|[A, C, E, B, D]|
+---+---------------+
+---+---------+
|id2| items2|
+---+---------+
| 1| [A, C]|
| 2| [D]|
| 3|[E, A, B]|
| 4|[B, D, C]|
| 5| [F, B]|
| 6| [G, E]|
+---+---------+
现在,crossJoin
两个数据帧,它为您提供 df1
与 df2
的笛卡尔积。然后,在 'items1'
上 groupby
并应用 udf
以获得 'result_array'
.
get_array_udf = F.udf(lambda x,y:[1.0 if set(z) < set(x) else 0.0 for z in y], ArrayType(FloatType()))
df = df1.crossJoin(df2)\
.groupby(['id1', 'items1']).agg(F.collect_list('items2').alias('items2'))\
.withColumn('result_array', get_array_udf('items1', 'items2')).drop('items2')
df.show()
这为您提供了输出,
+---+---------------+------------------------------+
|id1|items1 |result_array |
+---+---------------+------------------------------+
|1 |[E, A, C] |[1.0, 0.0, 0.0, 0.0, 0.0, 0.0]|
|0 |[B, C, D, E] |[0.0, 1.0, 0.0, 1.0, 0.0, 0.0]|
|4 |[A, C, E, B, D]|[1.0, 1.0, 1.0, 1.0, 0.0, 0.0]|
|3 |[E, G, A] |[0.0, 0.0, 0.0, 0.0, 0.0, 1.0]|
|2 |[F, A, E, B] |[0.0, 0.0, 1.0, 0.0, 1.0, 0.0]|
+---+---------------+------------------------------+