有没有办法释放 xarray.Dataset 的文件锁？

Question

我有一个使用 netcdf4.Dataset(fn, mode=a) 每 5 分钟增长一个 NetCDF 文件 fn 的进程。我还有一个使用xarray.Dataset 的NetCDF 文件的散景服务器可视化（我想保留它，因为它很方便）。

问题是 NetCDF-update-process 在尝试将新数据添加到 fn 时失败，如果它在我的散景服务器进程中通过

打开

ds = xarray.open_dataset(fn)

如果我使用选项autoclose

ds = xarray.open_dataset(fn, autoclose=True)

当 ds 在散景服务器应用程序中“打开”时使用其他进程更新 fn 可以正常工作，但是从 fn 提取时间片的散景图的更新变得非常滞后。

我的问题是：在使用xarray.Dataset时，是否有其他方法可以释放NetCDF文件的锁？

我不在乎 xarray.Dataset 的形状是否仅在重新加载整个散景服务器应用程序后才会持续更新。

谢谢！

这是一个最小的工作示例：

把它放到一个文件中并让它运行：

import time
from datetime import datetime

import numpy as np
import netCDF4

fn = 'my_growing_file.nc'

with netCDF4.Dataset(fn, 'w') as nc_fh:
    # create dimensions
    nc_fh.createDimension('x', 90)
    nc_fh.createDimension('y', 90)
    nc_fh.createDimension('time', None)

    # create variables
    nc_fh.createVariable('x', 'f8', ('x'))
    nc_fh.createVariable('y', 'f8', ('y'))
    nc_fh.createVariable('time', 'f8', ('time'))
    nc_fh.createVariable('rainfall_amount',
                         'i2',
                         ('time', 'y', 'x'),
                         zlib=False,
                         complevel=0,
                         fill_value=-9999,
                         chunksizes=(1, 90, 90))
    nc_fh['rainfall_amount'].scale_factor = 0.1
    nc_fh['rainfall_amount'].add_offset = 0

    nc_fh.set_auto_maskandscale(True)

    # variable attributes
    nc_fh['time'].long_name = 'Time'
    nc_fh['time'].standard_name = 'time'
    nc_fh['time'].units = 'hours since 2000-01-01 00:50:00.0'
    nc_fh['time'].calendar = 'standard'

for i in range(1000):
    with netCDF4.Dataset(fn, 'a') as nc_fh:
        current_length = len(nc_fh['time'])

        print('Appending to NetCDF file {}'.format(fn))
        print(' length of time vector: {}'.format(current_length))

        if current_length > 0:
            last_time_stamp = netCDF4.num2date(
                nc_fh['time'][-1],
                units=nc_fh['time'].units,
                calendar=nc_fh['time'].calendar)
            print(' last time stamp in NetCDF: {}'.format(str(last_time_stamp)))
        else:
            last_time_stamp = '1900-01-01'
            print(' empty file, starting from scratch')

        nc_fh['time'][i] = netCDF4.date2num(
            datetime.utcnow(),
            units=nc_fh['time'].units,
            calendar=nc_fh['time'].calendar)
        nc_fh['rainfall_amount'][i, :, :] = np.random.rand(90, 90)

    print('Sleeping...\n')
    time.sleep(3)

然后，转到例如IPython 并通过以下方式打开不断增长的文件：

ds = xr.open_dataset('my_growing_file.nc')

这将导致附加到 NetCDF 的进程失败，输出如下：

Appending to NetCDF file my_growing_file.nc
 length of time vector: 0
 empty file, starting from scratch
Sleeping...

Appending to NetCDF file my_growing_file.nc
 length of time vector: 1
 last time stamp in NetCDF: 2018-04-12 08:52:39.145999
Sleeping...

Appending to NetCDF file my_growing_file.nc
 length of time vector: 2
 last time stamp in NetCDF: 2018-04-12 08:52:42.159254
Sleeping...

Appending to NetCDF file my_growing_file.nc
 length of time vector: 3
 last time stamp in NetCDF: 2018-04-12 08:52:45.169516
Sleeping...

---------------------------------------------------------------------------
IOError                                   Traceback (most recent call last)
<ipython-input-17-9950ca2e53a6> in <module>()
     37 
     38 for i in range(1000):
---> 39     with netCDF4.Dataset(fn, 'a') as nc_fh:
     40         current_length = len(nc_fh['time'])
     41 

netCDF4/_netCDF4.pyx in netCDF4._netCDF4.Dataset.__init__()

netCDF4/_netCDF4.pyx in netCDF4._netCDF4._ensure_nc_success()

IOError: [Errno -101] NetCDF: HDF error: 'my_growing_file.nc'

如果使用

ds = xr.open_dataset('my_growing_file.nc', autoclose=True)

没有错误，但是通过xarray 的访问时间当然会变慢，这正是我的问题，因为我的仪表板可视化变得非常滞后。

我可以理解这可能不是xarray 的预期用途，如果需要，我将回退到netCDF4 提供的较低级别的接口（希望它支持并发文件访问，至少对于读取） )，但为了方便起见，我想保留xarray。

Answer 1

我在这里回答我自己的问题是因为我找到了一个解决方案，或者更好地说，一种解决 Python 中 NetCDF 文件锁定问题的方法。

一个好的解决方案是使用zarr 而不是 NetCDF 文件，如果您想在文件中持续增长数据集同时保持其打开状态，例如实时可视化。

幸运的是，xarray 现在还可以使用append_dim keyword argument 轻松地将数据附加到现有的 zarr 文件中，这要感谢a recently merged PR。

在我的问题中使用 zarr 而不是 NetCDF 的代码如下所示：

import dask.array as da
import xarray as xr
import pandas as pd
import datetime
import time

fn = '/tmp/my_growing_file.zarr'

# Creat a dummy dataset and write it to zarr
data = da.random.random(size=(30, 900, 1200), chunks=(10, 900, 1200))
t = pd.date_range(end=datetime.datetime.utcnow(), periods=30, freq='1s')
ds = xr.Dataset(
    data_vars={'foo': (('time', 'y', 'x'), data)},
    coords={'time': t},
)
#ds.to_zarr(fn, mode='w', encoding={'foo': {'dtype': 'int16', 'scale_factor': 0.1, '_FillValue':-9999}})
#ds.to_zarr(fn, mode='w', encoding={'time': {'_FillValue': -9999}})
ds.to_zarr(fn, mode='w')

# Append new data in smaller chunks
for i in range(100):
    print('Sleeping for 10 seconds...')
    time.sleep(10)

    data = 0.01 * i + da.random.random(size=(7, 900, 1200), chunks=(7, 900, 1200))
    t = pd.date_range(end=datetime.datetime.utcnow(), periods=7, freq='1s')
    ds = xr.Dataset(
        data_vars={'foo': (('time', 'y', 'x'), data)},
        coords={'time': t},
    )
    print(f'Appending 7 new time slices with latest time stamp {t[-1]}')
    ds.to_zarr(fn, append_dim='time')

然后您可以打开另一个 Python 进程，例如IPython 和做

 ds = xr.open_zarr('/tmp/my_growing_file.zarr/')   

一遍又一遍，而不会导致编写器进程崩溃。

我在这个例子中使用了 xarray 版本 0.15.0 和 zarr 版本 2.4.0。

一些补充说明：

请注意，此示例中的代码故意附加到小块中，这些小块不均匀地划分 zarr 文件中的块大小，以查看这对块的影响。根据我的测试，我可以说最初选择的 zarr 文件的块大小被保留了，这很棒！

另请注意，代码在追加时会生成警告，因为datetime64 数据已被xarray 编码并存储为整数，以符合 NetCDF 的 CF 约定。这也适用于 zarr 文件，但目前似乎没有自动设置 _FillValue。只要您的时间数据中没有NaT，这无关紧要。

免责声明：我还没有尝试过使用更大的数据集和长时间运行的文件增长过程，因此我无法评论最终的性能下降或如果 zarr 文件或其元数据以某种方式碎片化可能出现的其他问题这个过程。