Pyspark-как группировать и агрегировать с помощью RDD - PullRequest
Новая
       13

Pyspark-как группировать и агрегировать с помощью RDD

0 голосов
/ 29 апреля 2019

У меня есть фреймы данных 

df_1 = spark.createDataFrame(
[(1, [(1), (4), (2) ,(3)])],
("id", "transactions")
)

df_2 = spark.createDataFrame(
[([ (1),(2),(3)], 2.0) ],
("items", "cost")
)

df_1_rdd= df_1.rdd
df_2_rdd=df_2.rdd

У меня есть UDF, чтобы проверить, все ли элементы массива присутствуют в другом массиве 

@udf("boolean")
def contains_all(x, y):
    if x is not None and y is not None:
        return set(y).issubset(set(x))

Как группировать и агрегировать, используя rdd? 

costs=(df_1_rdd
.cartesian(df_2_rdd).groupBy("id", "transactions")
.agg(sum_(when(
   contains_all("transactions", "items"), col("support")
 )).alias("score")))
...