读取多个文件但在 python 中对一个文件执行摘要的优雅方式

Posted 2023-04-15

【中文标题】读取多个文件但在 python 中对一个文件执行摘要的优雅方式

【英文标题】：Elegant way to read multiple files but perform summary on one in python

【发布时间】：2019-10-23 07:25:47

【问题描述】：

我有多个文件，如下所示。我的任务是读取所有这些文件，合并它们并创建一个最终数据框。但是，一个文件 (Measurement_table_sep_13th.csv) 在用于合并之前必须进行汇总。太大了，我们总结一下，然后合并。

filenames = sorted(glob.glob('*.csv'))
filenames   # gives the below output

filenames = sorted(glob.glob('*.csv'))
for f in filenames:
   print(f)
   if f == 'Measurement_table_sep_13th.csv':
       df = spark.read.csv(f, sep=",",inferSchema=True, header=True)
       df = df.groupby("person_id","visit_occurrence_id").pivot("measurement_concept_id").agg(F.mean(F.col("value_as_number")), F.min(F.col("value_as_number")), F.max(F.col("value_as_number")),
                                            F.count(F.col("value_as_number")),F.stddev(F.col("value_as_number")),
                                            F.expr('percentile_approx(value_as_number, 0.25)').alias("25_pc"),
                                            F.expr('percentile_approx(value_as_number, 0.75)').alias("75_pc"))
   else:
       df = spark.read.csv(f, sep=",",inferSchema=True, header=True)

   try:
      JKeys = ['person_id', 'visit_occurrence_id'] if 'visit_occurrence_id' in df.columns else ['person_id']
      print(JKeys)
      df_final = df_final.join(df, on=JKeys, how='left')
      print("success in try")
   except:
      df_final = df
      print("success in except")

如您所见，我在合并之前对Measurement_table_sep_13th.csv 文件进行了总结，但是还有其他优雅高效的方法来编写它吗？

【问题讨论】：

为什么不将文件保存在不同的文件夹中？那么你可以完全避免 if-part

这本身似乎很有效。如果您想提高效率，执行需要多长时间？

@PV8 - 文件只能在同一个文件夹中

【参考方案1】：

如果你不想将一个文件保存在不同的文件夹中，你也可以直接用 glob 排除它：

紧随其后的是这篇文章： glob exclude pattern

files = glob.glob('files_path/[!_]*')

您可以使用它为除测量文件之外的所有文件运行 glob 函数，然后加入它。

那么你可以避免冗长的 if 代码。

看起来像（后面是这个帖子：Loading multiple csv files of a folder into one dataframe）：

files = glob.glob("[!M]*.csv")
dfs = [pd.read_csv(f, header=True, sep=";", inferShema=True) for f in files]

df2 = pd.concat(dfs,ignore_index=True)
df = spark.read.csv(f, sep=",",inferSchema=True, header=True)
df = df.groupby("person_id","visit_occurrence_id").pivot("measurement_concept_id").agg(F.mean(F.col("value_as_number")), F.min(F.col("value_as_number")), F.max(F.col("value_as_number")),
                                            F.count(F.col("value_as_number")),F.stddev(F.col("value_as_number")),
                                            F.expr('percentile_approx(value_as_number, 0.25)').alias("25_pc"),
                                            F.expr('percentile_approx(value_as_number, 0.75)').alias("75_pc"))
JKeys = ['person_id', 'visit_occurrence_id'] if 'visit_occurrence_id' in df.columns else ['person_id']
df_final = df(df2, on=JKeys, how='left')