Как эффективно повторяться через Pyspark DataFrame? ⇐ Python

[Anonymous]

У меня есть DataFrame, которая выглядит примерно так: < /p>



Материал < /th>
< /th> < /th>
< /th> < /th>
< /th> < /th>
/> refectquantity < /th>
< /tr>
< /thead>


< /td>
a1 < /td>
1300 < /td>
/>
a < /td>
a2 < /td>
1300 < /td>
td>>0.056
< /tr>

< /td>
< /td> < /td> < /td> < /td> < /td> < /td> < /td> < /td> < /td> < /td> < /td> /> 1300 < /td>
2.78
< /tr>

< /td>
b < /td>
1300 < /td>
td>1300.5 1300 < /td>
td>1300.5,5. />
b < /td>
b1 < /td>
1000 < /td>
1007 < /td>
< /tr>

b < /td>
b2
b < /td>
b2
b < /td>
b2
b < /td>
b2
b < /td> />1000
3.5


B
C
1000
9


c < /td>
c1 < /td>
800 < /td>
806.4
< /tr>
< /tbode>
< /table> < /div>
для каждого материала. Компоненты с самым низким уровнем при добавлении строки для каждого из них и выполняя расчет для нормализации требуемой квалификации для новых строк: requiredquantity / pactor * parent requiredquantity < / code>. The resulting DataFrame should look like this:



Material
Component
BatchSize
RequiredQuantity
< /tr>
< /thead>


a < /td>
a1 < /td>
1300 < /td>
1.0
< /tr>

1.0. /> a < /td>
a2 < /td>
1300 < /td>
0.056
< /tr>

< /td>
a3
< /td>

< /td>

< /td>
/> 1300 < /td>
2.78
< /tr>

< /td>
b < /td>
1300 < /td>
td>1300.5 1300 < /td>
td>1300.5,5. />
a < /td>
b1 < /td>
1300 < /td>
td>1309.6035
< /tr>

< /td>


< /td>


< /td>
< /tr>

< /td>


< /td>
/> 1300 < /td>
4.5.551755
< /tr>

a < /td>
c < /td>
1300 < /td>
>11.11.704. /> < /tr>

a < /td>
c1 < /td>
1300 < /td>
td>11.798136> < /tr>

b < /td> b < /td> b < /td> b < /td> b b b < /td> /> b1 < /td>
1000 < /td>
1007 < /td>
< /tr>

b < /td>
b2 < /td>
1000 < /td>
b2 < /td>
1000 < /td>
/>3.5,5
< /tr>

b < /td>
c < /td>
1000 < /td>
9 < /td>
< /tr>


< /td>
< /tr>
/> c1 < /td>
1000 < /td>
9.072ho />806.4
< /tr>
< /tbody>
< /table> < /div>
Я попытался написать рекурсивную функцию, которая работает, но является чрезвычайно медленным, занимая примерно 5 минут на материал. Это было бы хорошо для небольшого стола, но в нашем случае у нас есть почти 5000 различных материалов, каждый из которых имеет примерно 10 различных компонентов, поэтому потребуются недели, чтобы пройти через все это. Я надеюсь, что есть лучший способ справиться с этим.

Код: Выделить всё

def recurse_components(df, material):
if df.isEmpty():
return df

filtered_material = df.where(F.col("Material") == material)
batch_size = filtered_material.select("BatchSize").first()["BatchSize"]

component_list = (
filtered_material.select("Component").rdd.flatMap(lambda x: x).collect()
)

for component in component_list:
component_table = df.where(F.col("Material") == component)
if not component_table.isEmpty():
required_quantity = (
filtered_material.where(F.col("Component") == component)
.select("RequiredQuantity")
.first()["RequiredQuantity"]
)
recursive_call = recurse_components(df, component).withColumns(
{
"Material": F.lit(material),
"RequiredQuantity": F.col("RequiredQuantity")
* required_quantity
/ F.col("BatchSize"),
"BatchSize": F.lit(batch_size),
}
)
filtered_material = filtered_material.union(recursive_call)

return filtered_material

material_list = df.select("Material").distinct().rdd.flatMap(lambda x: x).collect()

extended_df = spark.createDataFrame([], df.schema)
for material in material_list:
extended_df = extended_df.union(recurse_components(df, material))

Любая помощь будет высоко оценена.

Подробнее здесь: https://stackoverflow.com/questions/796 ... -dataframe