2

问题:我有现有的 netCDF4 文件(大约 5000 个),(通常形状为 96x3712x3712)数据点(float32)。这些文件的第一个维度是时间(每天 1 个文件),第二个和第三个空间维度。目前,在第一个维度上进行切片(甚至是部分切片)将花费大量时间,原因如下:

  • netCDF 文件以 1x3712x3712 的块大小进行分块。在时间维度上切片基本上会读取整个文件。
  • 在所有较小的文件上循环(即使在多个进程中)也将花费大量时间。

我的目标:

  • 创建每月文件(约 2900x3712x3712)数据点
  • 优化它们以在时间维度上切片(块大小为 2900x1x1 或在空间维度上稍大)

其他需求:

  • 文件应可附加单个时间戳 (1x3712x3712),此更新过程应少于 15 分钟
  • 查询应该足够快:在不到一秒的时间内完成一个完整的切片(即 2900x1x1)==> 实际上没有那么多数据......
  • 最好在更新时文件应该可以被多个进程读取以供读取
  • 处理历史数据(其他 5000 个每日文件)最好在几周内完成。

我已经尝试了多种方法:

  • 连接netcdf文件并重新分块==>占用太多内存和太多时间......
  • 将它们从 pandas 写入 hdf 文件(使用 pytables)==> 创建一个具有巨大索引的宽表。这最终也会花费太多时间来读取,并且由于元数据限制,需要将数据集平铺在空间维度上。
  • 我的最后一种方法是使用 h5py 将它们写入 hdf5 文件:

这是创建单个月度文件的代码:

import h5py
import pandas as pd
import numpy as np

def create_h5(fps):
    timestamps=pd.date_range("20050101",periods=31*96,freq='15T') #Reference time period
    output_fp = r'/data/test.h5'
    try:
        f = h5py.File(output_fp, 'a',libver='latest')
        shape = 96*nodays, 3712, 3712
        d = f.create_dataset('variable', shape=(1,3712,3712), maxshape=(None,3712,3712),dtype='f', compression='gzip', compression_opts=9,chunks=(1,29,29))
        f.swmr_mode = True
        for fp in fps:
            try:
                nc=Dataset(fp)
                times = num2date(nc.variables['time'][:], nc.variables['time'].units)
                indices=np.searchsorted(timestamps, times)
                for j,time in enumerate(times):
                    logger.debug("File: {}, timestamp: {:%Y%m%d %H:%M}, pos: {}, new_pos: {}".format(os.path.basename(fp),time,j,indices[j]))
                    d.resize((indices[j]+1,shape[1],shape[2]))
                    d[indices[j]]=nc.variables['variable'][j:j+1]
                    f.flush()
            finally:
                nc.close()
    finally:
        f.close()
    return output_fp

我正在使用 HDF5 的最新版本来获得 SWMR 选项。fps 参数是日常 netCDF4 文件的文件路径列表。它在大约 2 小时内创建文件(在 ssd 上,但我发现创建文件主要受 CPU 限制),这是可以接受的。

我设置了压缩以将文件大小保持在限制范围内。我在没有的情况下进行了较早的测试,发现没有情况下的创建速度要快一些,但是压缩时切片所需的时间不会太长。H5py 自动将数据集分块为 1x116x116 块。

现在的问题是:在具有 RAID 6 设置的 NAS 上进行切片,需要大约 20 秒来切片时间维度,即使它位于单个块中......

我想,即使它在文件中的单个块中,因为我在一个循环中编写了所有值,它必须以某种方式分段(虽然不知道这个过程是如何工作的)。这就是为什么我尝试使用 HDF5 的 CML 工具将 h5repack 放入一个新文件中,使用相同的块但希望重新排序值,以便查询能够以更顺序的顺序读取值,但没有运气。尽管这个过程运行了 6 小时,但它对查询速度没有任何影响。

如果我的计算正确,读取一个块(2976x32x32)只有几 MB 大(11MB 未压缩,我认为仅比 1MB 多一点压缩)。这怎么可能需要这么长时间?我究竟做错了什么?如果有人可以了解幕后实际发生的事情,我会很高兴...

4

1 回答 1

3

块大小的影响

在最坏的情况下,读取和写入一个块可以被视为随机读/写操作。SSD 的主要优势是读取或写入小块数据的速度。HDD 在此任务中要慢得多(可以观察到 100 倍),NAS 甚至可能比 HDD 慢得多。

所以问题的解决方案将是更大的块大小。我系统上的一些基准测试(Core i5-4690)。

Example_1(块大小(1,29,29)=3,4 kB):

import numpy as np
import tables #needed for blosc
import h5py as h5
import time
import h5py_cache as h5c

def original_chunk_size():
    File_Name_HDF5='some_Path'
    #Array=np.zeros((1,3712,3712),dtype=np.float32)
    Array=np.random.rand(96,3712,3712)

    f = h5.File(File_Name_HDF5, 'a',libver='latest')
    f.swmr_mode = True
    nodays=1

    shape = 96*nodays, 3712, 3712
    d = f.create_dataset('variable', shape, maxshape=(None,3712,3712),dtype='f',chunks=(1,29,29),compression=32001,compression_opts=(0, 0, 0, 0, 9, 1, 1), shuffle=False)

    #Writing
    t1=time.time()
    for i in xrange(0,96*nodays):
        d[i:i+1,:,:]=Array

    f.close()
    print(time.time()-t1)

    #Reading
    f = h5.File(File_Name_HDF5, 'a',libver='latest')
    f.swmr_mode = True
    d=f['variable']

    for i in xrange(0,3712,29):
        for j in xrange(0,3712,29):
            A=np.copy(d[:,i:i+29,j:j+29])

    print(time.time()-t1)

结果(写/读):

固态硬盘:38s/54s

硬盘:40s/57s

NAS:252s/823s

在第二个示例中,我将使用 h5py_chache,因为我不想保持提供 (1,3712,3712) 的块。标准的 chunk-chache-size 只有 1 MB,因此必须对其进行更改,以避免对块进行多次读/写操作。 https://pypi.python.org/pypi/h5py-cache/1.0

Example_2(块大小 (96,58,58)=1,3 MB):

import numpy as np
import tables #needed for blosc
import h5py as h5
import time
import h5py_cache as h5c

def modified_chunk_size():
    File_Name_HDF5='some_Path'
    Array=np.random.rand(1,3712,3712)

    f = h5c.File(File_Name_HDF5, 'a',libver='latest', 
    chunk_cache_mem_size=6*1024**3)
    f.swmr_mode = True
    nodays=1

    shape = 96*nodays, 3712, 3712
    d = f.create_dataset('variable', shape, maxshape=(None,3712,3712),dtype='f',chunks=(96,58,58),compression=32001,compression_opts=(0, 0, 0, 0, 9, 1, 1), shuffle=False)

    #Writing
    t1=time.time()
    for i in xrange(0,96*nodays):
        d[i:i+1,:,:]=Array

    f.close()
    print(time.time()-t1)

    #Reading
    f = h5c.File(File_Name_HDF5, 'a',libver='latest', chunk_cache_mem_size=6*1024**3) #6 GB chunk chache
    f.swmr_mode = True
    d=f['variable']

    for i in xrange(0,3712,58):
        for j in xrange(0,3712,58):
            A=np.copy(d[:,i:i+58,j:j+58])

    print(time.time()-t1)

结果(写/读):

固态硬盘:10s/16s

硬盘:10s/16s

NAS:13s/20s

通过最小化 api 调用(读取和写入更大的块块)可以进一步提高读/写速度。

我也不想提她的压缩方法。Blosc 可以实现高达 1GB/s 的吞吐量(CPU 瓶颈) gzip 速度较慢,但​​提供更好的压缩比。

d = f.create_dataset('variable', shape, maxshape=(None,3712,3712),dtype='f',chunks=(96,58,58),compression='gzip', compression_opts=3)

20s/30s 文件大小:101 MB

d = f.create_dataset('variable', shape, maxshape=(None,3712,3712),dtype='f',chunks=(96,58,58),compression='gzip', compression_opts=6)

50s/58s 文件大小:87 MB

d = f.create_dataset('variable', shape, maxshape=(None,3712,3712),dtype='f',chunks=(96,58,58),compression='gzip', compression_opts=9)

50s/60s 文件大小:64 MB

现在是一个整月(30天)的基准。写法有点优化,用(96,3712,3712)写。

def modified_chunk_size():
    File_Name_HDF5='some_Path'

    Array_R=np.random.rand(1,3712,3712)
    Array=np.zeros((96,3712,3712),dtype=np.float32)
    for j in xrange(0,96):
        Array[j,:,:]=Array_R

    f = h5.File(File_Name_HDF5, 'a',libver='latest')
    f.swmr_mode = True
    nodays=30

    shape = 96, 3712, 3712
    d = f.create_dataset('variable', shape, maxshape=(None,3712,3712),dtype='f',chunks=(96,58,58),compression=32001,compression_opts=(0, 0, 0, 0, 9, 1, 1), shuffle=False)

    #Writing
    t1=time.time()
    for i in xrange(0,96*nodays,96):
        d[i:i+96,:,:]=Array
        d.resize((d.shape[0]+96,shape[1],shape[2]))

    f.close()
    print(time.time()-t1)

    #Reading
    f = h5.File(File_Name_HDF5, 'a',libver='latest')
    f.swmr_mode = True
    d=f['variable']
    for i in xrange(0,3712,58):
        for j in xrange(0,3712,58):
            A=np.copy(d[:,i:i+58,j:j+58])

    print(time.time()-t1)

133s/301s 带块

432s/684s with gzip compression_opts=3

我在访问 NAS 上的数据时遇到了同样的问题。我希望这有帮助...

于 2017-07-07T01:32:38.363 回答