<jupyter_start><jupyter_text>Xarray to Zarr on GCS with CamHD Data ExampleHere we build an Xarray Dataset from Dask delayed functions, and save the images to a GCS zarr group using xarray's to_zarr() functionality. Processing takes place on the workers and memory pressure is very low because all the images do not need to be loaded at the same time.  Imports<jupyter_code>%matplotlib inline
import pycamhd as camhd
import numpy as np
import matplotlib.pyplot as plt
import pandas as pd
import xarray as xr<jupyter_output><empty_output><jupyter_text>Create a list of frames to analyze using the dbcamhd.json database<jupyter_code>dbcamhd = pd.read_json('dbcamhd.json', orient="records", lines=True)
dbcamhd.tail()
fileindex = 2064
filename = dbcamhd.filename[fileindex]
frame_count = dbcamhd.frame_count[fileindex]
n_images = 1000
frame_numbers = np.linspace(750,frame_count-6000, n_images, dtype=np.int64())
filename
frame_numbers[0:10]<jupyter_output><empty_output><jupyter_text>Create timestamps for frames<jupyter_code>from datetime import datetime
timestamps = []
for i in range(len(frame_numbers)):
    timestamps.append(datetime.fromtimestamp(dbcamhd.timestamp[fileindex] + frame_numbers[i]/29.95))
timestamps[0:5]<jupyter_output><empty_output><jupyter_text>Create Xarray Dataset out of Dask delayed functions<jupyter_code>from dask import delayed, compute
import dask.array as da
delayed_frames = []
moov_atom = camhd.get_moov_atom(filename)
for frame_number in frame_numbers:
    delayed_frames.append(da.from_delayed(
                            delayed(camhd.get_frame)(filename, frame_number, 'rgb24', moov_atom),
                            shape=(1080, 1920, 3), dtype=np.uint8)[None,:,:,:])
delayed_frames[0]
all_data = da.concatenate(delayed_frames, axis=0)
all_data
ds = xr.DataArray(all_data, dims=['time', 'y', 'x', 'channel'],
                  coords={'time': timestamps}
                 ).to_dataset(name='video')
ds<jupyter_output><empty_output><jupyter_text>Start a KubeCluster<jupyter_code>from dask_kubernetes import KubeCluster
cluster = KubeCluster(n_workers=32)
cluster
from dask import delayed, compute
from dask.distributed import Client
client = Client(cluster)
client<jupyter_output><empty_output><jupyter_text>Use to_zarr() to save to a GCS bucket<jupyter_code>import google.auth
token, project = google.auth.default()
import zarr
gcsstore = zarr.storage.GCSStore('rte-pangeo-data', 'test1.zarr', client_kwargs={'project': 'pangeo-198314', 'credentials': token})
%%time
ds.to_zarr(gcsstore)<jupyter_output><empty_output><jupyter_text>Open and compute on data in GCS bucket<jupyter_code>%%time
del ds
ds = xr.open_zarr(gcsstore)
ds
%%time
mean_image = ds.video.mean(dim='time').load()
mean_image.astype('i8').plot.imshow();<jupyter_output><empty_output><jupyter_text>Use to_zarr() to save to Azure Blob Storage<jupyter_code>import zarr
import xarray as xr
from azure_credentials import account_name, account_key
absstore = zarr.storage.ABSStore('rte-pangeo-data', 'test1.zarr', account_name, account_key)
absstore.rmdir()
%%time
ds.to_zarr(absstore)<jupyter_output><empty_output><jupyter_text>Open and compute on data Azure Blob container<jupyter_code>%%time
del ds
ds = xr.open_zarr(absstore)
ds
%%time
mean_image = ds.video.mean(dim='time').load()
mean_image.astype('i8').plot.imshow();<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10918854415274463, 'fraction_numerical': 0.015513126491646777, 'mean_word_length': 6.501118568232662, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '195315', 'license': 'MIT', 'n_tokens_mistral': 1198, 'n_tokens_neox': 1161, 'n_words': 266, 'path': 'tim/xarray_tozarr.ipynb', 'repo_name': 'tjcrone/rte-camhd', 'size': '3352'}