Schreiben von Xarray-Multiindexdaten in Blöcken
Ich versuche, einen großen multidimentionalen Datensatz effizient umzustrukturieren. Nehmen wir an, ich habe im Laufe der Zeit eine Reihe von Bildern mit Fernerkundung mit einer Reihe von Bändern mit den Koordinaten xy für die Pixelposition, der Zeit für die Zeit der Bildaufnahme und dem Band für verschiedene gesammelte Daten.
In meinem Anwendungsfall nehmen wir an, dass die xarray-Koordinatenlängen ungefähr x (3000), y (3000), Zeit (10) mit Bändern (40) von Gleitkommadaten sind. Also mehr als 100 GB Daten.
Ich habe versucht, anhand dieses Beispiels zu arbeiten, aber ich habe Probleme, es in diesen Fall zu übersetzen.
Beispiel für einen kleinen Datensatz
HINWEIS: Die tatsächlichen Daten sind viel größer als in diesem Beispiel.
import numpy as np
import dask.array as da
import xarray as xr
nrows = 100
ncols = 200
row_chunks = 50
col_chunks = 50
data = da.random.random(size=(1, nrows, ncols), chunks=(1, row_chunks, col_chunks))
def create_band(data, x, y, band_name):
return xr.DataArray(data,
dims=('band', 'y', 'x'),
coords={'band': [band_name],
'y': y,
'x': x})
def create_coords(data, left, top, celly, cellx):
nrows = data.shape[-2]
ncols = data.shape[-1]
right = left + cellx*ncols
bottom = top - celly*nrows
x = np.linspace(left, right, ncols) + cellx/2.0
y = np.linspace(top, bottom, nrows) - celly/2.0
return x, y
x, y = create_coords(data, 1000, 2000, 30, 30)
src = []
for time in ['t1', 't2', 't3']:
src_t = xr.concat([create_band(data, x, y, band) for band in ['blue', 'green', 'red', 'nir']], dim='band')\
.expand_dims(dim='time')\
.assign_coords({'time': [time]})
src.append(src_t)
src = xr.concat(src, dim='time')
print(src)
<xarray.DataArray 'random_sample-5840d8564d778d573dd403f27c3f47a5' (time: 3, band: 4, y: 100, x: 200)>
dask.array<concatenate, shape=(3, 4, 100, 200), dtype=float64, chunksize=(1, 1, 50, 50), chunktype=numpy.ndarray>
Coordinates:
* x (x) float64 1.015e+03 1.045e+03 1.075e+03 ... 6.985e+03 7.015e+03
* band (band) object 'blue' 'green' 'red' 'nir'
* y (y) float64 1.985e+03 1.955e+03 1.924e+03 ... -984.7 -1.015e+03
* time (time) object 't1' 't2' 't3'
Umstrukturiert - gestapelt und transponiert
Ich muss die Ausgabe von Folgendem speichern:
print(src.stack(sample=('y','x','time')).T)
<xarray.DataArray 'random_sample-5840d8564d778d573dd403f27c3f47a5' (sample: 60000, band: 4)>
dask.array<transpose, shape=(60000, 4), dtype=float64, chunksize=(3600, 1), chunktype=numpy.ndarray>
Coordinates:
* band (band) object 'blue' 'green' 'red' 'nir'
* sample (sample) MultiIndex
- y (sample) float64 1.985e+03 1.985e+03 ... -1.015e+03 -1.015e+03
- x (sample) float64 1.015e+03 1.015e+03 ... 7.015e+03 7.015e+03
- time (sample) object 't1' 't2' 't3' 't1' 't2' ... 't3' 't1' 't2' 't3'
Ich hoffe, dass ich dask und xarray verwenden kann, um das Ergebnis in Blöcken auf die Festplatte zu schreiben, auf die open_mfdataset zugreifen kann . Parkett scheint eine gute Option zu sein, aber ich kann nicht herausfinden, wie man es in Stücken schreibt (src ist zu groß, um es im Speicher zu speichern).
@dask.delayed
def stacker(data):
return data.stack(sample=('y','x','time')).T.to_pandas()
stacker(src).to_parquet('out_*.parquet')
def stack_write(data):
data.stack(sample=('y','x','time')).T.to_pandas().to_parquet('out_*.parquet')
return None
stack_write(src)
An dieser Stelle hoffe ich nur auf ein paar gute Ideen. Vielen Dank!
Antworten
Ich habe hier eine Lösung (https://github.com/pydata/xarray/issues/1077#issuecomment-644803374) zum Schreiben von mehrfach indizierten Datensätzen in eine Datei.
Sie müssen das Dataset manuell in ein Formular "codieren", das als netCDF geschrieben werden kann. Und dann "dekodieren", wenn Sie es zurücklesen.
import numpy as np
import pandas as pd
import xarray as xr
def encode_multiindex(ds, idxname):
encoded = ds.reset_index(idxname)
coords = dict(zip(ds.indexes[idxname].names, ds.indexes[idxname].levels))
for coord in coords:
encoded[coord] = coords[coord].values
shape = [encoded.sizes[coord] for coord in coords]
encoded[idxname] = np.ravel_multi_index(ds.indexes[idxname].codes, shape)
encoded[idxname].attrs["compress"] = " ".join(ds.indexes[idxname].names)
return encoded
def decode_to_multiindex(encoded, idxname):
names = encoded[idxname].attrs["compress"].split(" ")
shape = [encoded.sizes[dim] for dim in names]
indices = np.unravel_index(encoded.landpoint.values, shape)
arrays = [encoded[dim].values[index] for dim, index in zip(names, indices)]
mindex = pd.MultiIndex.from_arrays(arrays)
decoded = xr.Dataset({}, {idxname: mindex})
for varname in encoded.data_vars:
if idxname in encoded[varname].dims:
decoded[varname] = (idxname, encoded[varname].values)
return decoded
Dies ist im Moment nicht die Lösung, sondern eine Version Ihres Codes, die so modifiziert wurde, dass sie leicht reproduzierbar ist, wenn andere versuchen möchten, dieses Problem zu lösen:
Das Problem liegt in der stackOperation ( concatenated.stack(sample=('y','x','time')). In diesem Schritt nimmt der Speicher weiter zu und der Prozess ist killed.
Das concatenatedObjekt ist ein "Dask-Backed" xarray.DataArray. Wir konnten also erwarten, dass die stackOperation von Dask träge durchgeführt wird. Warum ist der Prozess killedin diesem Schritt?
2 Möglichkeiten für das, was hier passiert:
Die
stackOperation wird zwar von Dask träge ausgeführt, aber da die Daten sehr groß sind, ist selbst der für Dask erforderliche Mindestspeicher zu großDie
stackOperation wird NICHT von Dask unterstützt
import numpy as np
import dask.array as da
import xarray as xr
from numpy.random import RandomState
nrows = 20000
ncols = 20000
row_chunks = 500
col_chunks = 500
# Create a reproducible random numpy array
prng = RandomState(1234567890)
numpy_array = prng.rand(1, nrows, ncols)
data = da.from_array(numpy_array, chunks=(1, row_chunks, col_chunks))
def create_band(data, x, y, band_name):
return xr.DataArray(data,
dims=('band', 'y', 'x'),
coords={'band': [band_name],
'y': y,
'x': x})
def create_coords(data, left, top, celly, cellx):
nrows = data.shape[-2]
ncols = data.shape[-1]
right = left + cellx*ncols
bottom = top - celly*nrows
x = np.linspace(left, right, ncols) + cellx/2.0
y = np.linspace(top, bottom, nrows) - celly/2.0
return x, y
x, y = create_coords(data, 1000, 2000, 30, 30)
bands = ['blue', 'green', 'red', 'nir']
times = ['t1', 't2', 't3']
bands_list = [create_band(data, x, y, band) for band in bands]
src = []
for time in times:
src_t = xr.concat(bands_list, dim='band')\
.expand_dims(dim='time')\
.assign_coords({'time': [time]})
src.append(src_t)
concatenated = xr.concat(src, dim='time')
print(concatenated)
# computed = concatenated.compute() # "computed" is ~35.8GB
stacked = concatenated.stack(sample=('y','x','time'))
transposed = stacked.T
Man kann versuchen, die Werte von nrowsund zu ändern ncols, um die Größe von zu variieren concatenated. Und für die Leistung könnten / sollten wir das auch variieren chunks.
Hinweis: Ich habe es sogar versucht
concatenated.to_netcdf("concatenated.nc")
concatenated = xr.open_dataarray("concatenated.nc", chunks=10)
Dies dient dazu, sicherzustellen, dass es sich um ein von Dask unterstütztes DataArray handelt, und um auch die Chunks anpassen zu können. Ich habe verschiedene Werte ausprobiert für chunks: aber immer zu wenig Speicher.