python - 使用 pandas 和 dask 合并具有不同模式的 Parquet 文件

Question

我有一个包含大约 1000 个文件的 parquet 目录，并且架构不同。我想通过文件重新分区将所有这些文件合并到最佳数量的文件中。我使用带有 pyarrow 的 pandas 从目录中读取每个分区文件并将所有数据帧连接起来并将其写入一个文件。

使用这种方法，当数据量增加时，我会遇到内存问题并被杀死。所以我选择了另一种方法来完成这个过程。

我首先读取了一堆文件，使用 concat 合并并写入新的 parquet 目录。同样，第二次，我读取了第二批文件，将其连接为一个数据帧，并从第二个合并的数据帧中获取了一条记录。现在我有一个来自第二个合并数据帧的记录，我再次从文件中读取第一个合并数据帧并将其与第二个合并数据帧的记录合并。然后我使用 dask to_parquet, append 功能将新文件添加到该 parquet 文件夹。

它是一个有效的 parquet 文件吗？当我们从这个 parquet 读取数据时，我会得到所有的列，比如 parquet schema evolution 吗？它会类似于 spark merge schema 吗？

更新:

sample.parquet - contains 1000 part files

def read_files_from_path(inputPath):
   return {"inputPath": ["part-001","part-002",...,"part-100"]}


def mergeParquet(list_of_files,output_path)
   dfs_list = []
   for i in range:
      df = pd.read_parquet(i, engine='pyarrow')
      dfs_list.append(df)
   df = pd.concat(dfs_list,axis=0,sort=True)
   df_sample_record_df = df[2:3]

   if os.path.exists(output_path + '/_metadata'):
      files_in_output_path = getFiles(output_path)
      for f in files_in_output_path:
         temp_df = pd.read_parquet(f, engine='pyarrow')
         temp_combine_df = pd.concat(temp_df,df_sample_record_df) 
         temp_combine_df.repartition(partition_size="128MB") \
                .to_parquet(output_path+"/tmp",engine='pyarrow',
                            ignore_divisions=True,append=True)
         os.remove(output_path+"/"+each_file)
   return df

def final_write_parquet(df,output_path):
   if os.path.exists(output_path+"/tmp"):
      df.repartition(partition_size="128MB")\
              .to_parquet(output_path+str(self.temp_dir),engine='pyarrow',
                            ignore_divisions=True,append=True)
      files = os.listdir(output_path + "/tmp")
      for f in files:
         shutil.move(output_path+"/tmp"+"/"+f, output_path)
         shutil.rmtree(output_path+"/tmp")
   else:
      df.repartition(partition_size="128MB")\
                .to_parquet(output_path, engine='pyarrow', append=False)


if __name__ == "__main__":
   files_dict = read_files_from_path(inputPath)
   number_of_batches = 1000/500    # total files/batchsize
   for sub_file_names in np.array_split(files_dict[0], num_parts):
      paths = [os.path.join(root_dir, file_name) for file_name in sub_file_names]
      mergedDF = parquetMerge(paths)
      final_write_parquet(megedDF,outputPath)

Answer 1

Dask 数据帧假定所有分区都具有相同的模式(列名和数据类型)。如果您想混合具有几乎相同模式的不同数据集，那么您将需要手动处理。 Dask DataFrame 今天在这里不提供自动化支持。