python - 将许多巨大的 CSV 文件中的 XYZ 坐标有效地排序为小图 block

Question

我的问题和目标:

我有许多巨大的 (1-10GB) CSV 文件，其中包含数十亿个 XYZ 坐标。我需要将它们插值到规则网格中，相当有效(不需要在 super 计算机上花费数周的计算时间)。

我的方法:

当处理几个较小的文件时，我将所有文件加载到一个 Pandas 数据帧中:

file_names = [files for files in os.listdir(my_path) if files.endswith(".csv")]  
data_list = []
for file_name in file_names:    
  df = pd.read_csv(my_path + "\\" + file_name, header=None, names=["x","y","z"], dtype=np.float32)
  data_list.append(df)    
frame = pd.concat(data_list)

之后，我得到frame[x].min()等坐标的最小/最大，并计算我想要将数据分解成的图 block 范围以简化插值:

tile_extents = [min_x - (min_x%tile_size), max_x - (max_x%tile_size) + tile_size, min_y - (min_y%tile_size), max_y - (max_y%tile_size) + tile_size]
for i in range(int((tile_extents[1]-tile_extents[0])/tile_size)):
  for j in range(int((tile_extents[3]-tile_extents[2])/tile_size)):
    current_tile_extent = [tile_extents[0]+i*tile_size, tile_extents[0]+i*tile_size+tile_size, tile_extents[2]+j*tile_size, tile_extents[2]+j*tile_size+tile_size]
    current_frame = frame[(frame["x"]>current_tile_extent[0]) & (frame["x"]<current_tile_extent[1]) & (frame["y"]>current_tile_extent[2]) & (frame["y"]<current_tile_extent[3])]

我可以将其保存到 CSV 文件中，然后对这些小子集运行插值。 但是如果我有 100GB 或更多 XYZ 坐标该怎么办？如何订购以便我可以将它们保存到图 block 中？

Answer 1

pandas.read_csv()可以以 block 的形式工作(请参阅chunksize参数)。

以下代码首先读取所有文件以查找 x 和 y 的最小值和最大值，然后计算图 block 并为每个图 block 创建一个文件 ( >tiles\i,j.csv)。然后，它再次读取所有文件，这次将行写入适当的图 block 文件中。
请务必根据您的需要调整chunksize!

我用一些假数据对此进行了测试，看起来不错:

import os
import numpy as np
import pandas as pd

def get_df_in_chunks(file_names, **kwargs):
    """Read all files in chunks."""
    # Set default chunk size if not specified.
    kwargs['chunksize'] = kwargs.get('chunksize', 100000)
    for file_name in file_names:
        df = pd.read_csv(file_name, **kwargs)
        yield from iter(df)

my_path = '.'
file_names = [my_path + "\\" + file for file in os.listdir(my_path) if file.endswith(".csv")]
print('file_names: {}'.format(file_names))

# Read all files to determine the global min and max values.
min_x = +np.inf
max_x = -np.inf
min_y = +np.inf
max_y = -np.inf
for chunk in get_df_in_chunks(file_names, header=None, dtype=np.float32,
                              usecols=["x","y"], names=["x","y"]):
    print('  - chunk: {}'.format(chunk.shape))
    min_x = min(min_x, chunk["x"].min())
    max_x = max(max_x, chunk["x"].max())
    min_y = min(min_y, chunk["y"].min())
    max_y = max(max_y, chunk["y"].max())
print("x min/max:", min_x, max_x)
print("y min/max:", min_y, max_y)

# Calculate tile extents.
tile_size = 1.5
tile_limits = [min_x - (min_x%tile_size),
               max_x - (max_x%tile_size) + tile_size,
               min_y - (min_y%tile_size),
               max_y - (max_y%tile_size) + tile_size]
print("tile_size:", tile_size)
print("tile_limits:", tile_limits)

if not os.path.exists(my_path + "\\tiles"):
    os.mkdir(my_path + "\\tiles")

tile_files = []
try:
    # Open all tile files.
    for i in range(int(round(tile_limits[1]-tile_limits[0])/tile_size)):
        for j in range(int(round(tile_limits[3]-tile_limits[2])/tile_size)):
            tile_extent = [tile_limits[0] + i*tile_size,
                           tile_limits[0] + i*tile_size + tile_size,
                           tile_limits[2] + j*tile_size,
                           tile_limits[2] + j*tile_size + tile_size]
            f = open(my_path + "\\tiles\\{},{}.csv".format(i,j), 'w')
            tile_files.append((f, tile_extent))

    for chunk in get_df_in_chunks(file_names, header=None, dtype=np.float32,
                                  usecols=["x","y","z"], names=["x","y","z"]):
        # For each tile, write all relevant rows into the corresponding tile file.
        for f, tile_extent in tile_files:
            # Option 1: boolean condition.
            frame = chunk[(chunk["x"] >= tile_extent[0]) &
                          (chunk["x"] <  tile_extent[1]) &
                          (chunk["y"] >= tile_extent[2]) &
                          (chunk["y"] <  tile_extent[3])]
            # Option 2: query using numexpr. Could be faster, but wasn't in my test.
            #frame = chunk.query('x >= {} & x < {} & y >= {} & y < {}'.format(*tile_extent))

            ##print('file: {} tile_extent: {} frame: shape={}'.format(f.name, tile_extent, frame.shape))
            frame.to_csv(f, header=None, index=False)
finally:
    # Close all tile files.
    for f, _ in tile_files:
        f.close()