根据条件即将到来的表单列表总结价值和dict

Question

我在列表中有12个时期，这些列表并未根据所选产品具有更多。 另外，我有一个dict，其键和产品作为值列表。

{
    "20191": ["prod1","prod2","prod3"],
    "20192": ["prod2","prod3"],
    "20193": ["prod2"]
}

II需要根据周期选择数据并计算各个期间的总和。

sample_data

周期	产品	金额
20191	pod1	30
20192	pod1	30
20191	prod2	20
20191	prod3	60
20193	pod1 pod1	30
20193	prod2	30

输出

期	产品	金额
20191	pod1	110
20192		0
20192 20193	prod3	30

基本上，对于每个时期，从dict，仅选择这些时期总结一下。

我的代码需要大量时间：

list_series = []
df = spark.read.csv(path,header=True)
periods = df.select("period").distinct().collect()
for period in periods:
  df1 = df.filter(f"period = {period}").filter(F.col("product").isin(dict["period"]).groupBy("priod","product").agg(F.sum("Amount").alias("Amount")
  list_series.append(df1)
dataframe = reduce(DataFrame.unionAll,list_series)

有什么办法，我可以修改并提高性能？

Answer 1

解决方案

将输入字典变义为单元列表，然后创建一个新的Spark DataFrame，称为efferters，然后JOIN与原始数据框一起使用列的原始 ocage> ocipers 和产品，然后groupby 期间和汇总金额使用sum

d = [(i, k) for k, v in dct.items() for i in v]
filters = spark.createDataFrame(d, schema=['product', 'period'])

(
    df
    .join(filters, on=['period', 'product'], how='right')
    .groupby('period')
    .agg(F.sum('amount').alias('amount'))
    .fillna(0)
)

结果

+------+------+
|period|amount|
+------+------+
| 20191|   110|
| 20192|     0|
| 20193|    30|
+------+------+

Answer 2

通过以下输入：

df = spark.createDataFrame(
    [('20191', 'prod1', 30),
     ('20192', 'prod1', 30),
     ('20191', 'prod2', 20),
     ('20191', 'prod3', 60),
     ('20193', 'prod1', 30),
     ('20193', 'prod2', 30)],
    ['period', 'product', 'amount'])

periods = ["20191", "20192", "20193"]
period_products = {
    "20191": ["prod1","prod2","prod3"],
    "20192": ["prod2","prod3"],
    "20193": ["prod2"]
}

要使您的脚本更具性能，您将需要删除从一个人创建多个DF的步骤，然后将它们全部重新组合在一起。在一个数据框架中执行此操作而无需分开。

您可以在python中创建过滤器条件（在联接应添加性能提升之前过滤器），将其提供给过滤器功能，然后汇总。

conds = [f"((period = '{p}') and (product ='{prod}'))" for p in periods for prod in period_products[p]]
cond = ' or '.join(conds)

df_periods = spark.createDataFrame(
    [(p, i) for p in periods for i in period_products[p]],
    ['period', 'product']
)

df = (df_periods
    .join(df.filter(cond), ['period', 'product'], 'left')
    .groupBy('period', 'product')
    .agg(F.sum('amount').alias('amount'))
)

df.show()
# +------+-------+------+
# |period|product|amount|
# +------+-------+------+
# | 20191|  prod2|    20|
# | 20191|  prod1|    30|
# | 20191|  prod3|    60|
# | 20193|  prod2|    30|
# | 20192|  prod2|  null|
# | 20192|  prod3|  null|
# +------+-------+------+