将通过 FOR 循环函数输出的多个数据帧合并为一个数据帧

Posted 2023-04-15

【中文标题】将通过 FOR 循环函数输出的多个数据帧合并为一个数据帧

【英文标题】：Merge multiple dataframes outputted via a FOR loop function into one single dataframe

【发布时间】：2019-05-05 04:55:21

【问题描述】：

我有一个 FOR 循环函数，它遍历表和列 (zip) 的列表以获取最小值和最大值。每个组合的输出都是分开的，而不是一个单独的数据帧/表。有没有办法将 FOR 循环的结果组合成函数内的一个最终输出？

from pyspark.sql import functions as f

def minmax(tables, cols):
    for table, column in zip(tables, cols):
        minmax = spark.table(table).where(col(column).isNotNull()).select(f.lit(table).alias("table"), f.lit(column).alias("col"), min(col(column)).alias("min"), 
        max(col(column)).alias("max"))
        minmax.show()

tables = ["sales_123", "sales_REW"]
cols = ["costs", "price"]

minmax(tables, cols)

函数输出：

+---------+-----+---+---+
|    table|  col|min|max|
+---------+-----+---+---+
|sales_123|costs|  0|400|
+---------+-----+---+---+

+----------+-----+---+---+
|     table|  col|min|max|
+----------+-----+---+---+
|sales_REW |price|  0|400|
+----------+-----+---+---+

期望的输出：

+---------+-----+---+---+
|    table|  col|min|max|
+---------+-----+---+---+
|sales_123|costs|  0|400|
|sales_REW|price|  0|400|
+---------+-----+---+---+

【参考方案1】：

将所有数据框放入一个列表中，并在for循环之后进行并集：

from functools import reduce
from pyspark.sql import functions as f
from pyspark.sql import DataFrame

def minmax(tables, cols):

    dfs = []        
    for table, column in zip(tables, cols):
        minmax = spark.table(table).where(col(column).isNotNull()).select(f.lit(table).alias("table"), f.lit(column).alias("col"), min(col(column)).alias("min"), max(col(column)).alias("max"))
        dfs.append(minmax)
    df = reduce(DataFrame.union, dfs)

请注意，所有涉及的数据帧的列顺序必须相同（就像这里的情况一样）。否则，这可能会产生意想不到的结果。

【讨论】：

@Shaido 我如何准备表格和列来测试代码？

@PIG：如果你只是想测试它，你可以简单地创建一个 Spark 数据帧列表（从 csvs 读取，或者从 pandas 数据帧创建一些：***.com/questions/43751509/…），然后循环这些数据。

@Shaido 我正在尝试传递minmax(df1, cols)，其中 df1 是我的 spark 数据框 & cols = df1.columns，它表示列不可迭代

@Abhi：如果你只有一个数据框，那么你不应该使用上面的解决方案。但是，您想做的事情会因您实际想做的事情而有很大不同，可能类似于：***.com/questions/60534796/… 或 ***.com/questions/45382329/…

@Shaido 是的！只需运行df.select(*[f.mean(c).alias(c) for c in df.columns])。你能帮忙处理百分位数吗，我有一个 ID（唯一）列，类别（5 个不同的值）列，5 个数字列。百分位数 - 每个数字列需要 25%、50%、75%、100%。将循环运行 5 个类别