<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
Step2:  Profile array copy via dask threaded scheduler
This notebook profiles a very simple array copy operation, using synthetic data.
Step3:  NumPy arrays
Step4:  Zarr arrays (in-memory)
Step5:  Without the dask lock, we get better CPU utilisation. 
HDF5 datasets (in-memory)
Step6:  Bcolz carrays (in-memory)
<ASSISTANT_TASK:>
Python Code:
import sys
sys.path.insert(0, '..')
import zarr
print('zarr', zarr.__version__)
from zarr import blosc
import numpy as np
import h5py
import bcolz
# don't let bcolz use multiple threads internally, we want to 
# see whether dask can make good use of multiple CPUs
bcolz.set_nthreads(1)
import multiprocessing
import dask
import dask.array as da
from dask.diagnostics import Profiler, ResourceProfiler, CacheProfiler
from dask.diagnostics.profile_visualize import visualize
from cachey import nbytes
import bokeh
from bokeh.io import output_notebook
output_notebook()
import tempfile
import operator
from functools import reduce
from zarr.util import human_readable_size
def h5fmem(**kwargs):
    Convenience function to create an in-memory HDF5 file.
    # need a file name even tho nothing is ever written
    fn = tempfile.mktemp()
    # file creation args
    kwargs['mode'] = 'w'
    kwargs['driver'] = 'core'
    kwargs['backing_store'] = False
    # open HDF5 file
    h5f = h5py.File(fn, **kwargs)
    return h5f
def h5d_diagnostics(d):
    Print some diagnostics on an HDF5 dataset.
    
    print(d)
    nbytes = reduce(operator.mul, d.shape) * d.dtype.itemsize
    cbytes = d._id.get_storage_size()
    if cbytes > 0:
        ratio = nbytes / cbytes
    else:
        ratio = np.inf
    r = '  compression: %s' % d.compression
    r += '; compression_opts: %s' % d.compression_opts
    r += '; shuffle: %s' % d.shuffle
    r += '\n  nbytes: %s' % human_readable_size(nbytes)
    r += '; nbytes_stored: %s' % human_readable_size(cbytes)
    r += '; ratio: %.1f' % ratio
    r += '; chunks: %s' % str(d.chunks)
    print(r)
 
def profile_dask_copy(src, dst, chunks, num_workers=multiprocessing.cpu_count(), dt=0.1, lock=True):
    dsrc = da.from_array(src, chunks=chunks)
    with Profiler() as prof, ResourceProfiler(dt=dt) as rprof:
        da.store(dsrc, dst, num_workers=num_workers, lock=lock)
    visualize([prof, rprof], min_border_top=60, min_border_bottom=60)
    
Explanation: Profile array copy via dask threaded scheduler
This notebook profiles a very simple array copy operation, using synthetic data.
End of explanation
# a1 = np.arange(400000000, dtype='i4')
a1 = np.random.normal(2000, 1000, size=200000000).astype('u2')
a1
human_readable_size(a1.nbytes)
a2 = np.empty_like(a1)
chunks = 2**20,  # 4M
%time a2[:] = a1
profile_dask_copy(a1, a2, chunks, lock=True, dt=.01)
profile_dask_copy(a1, a2, chunks, lock=False, dt=.01)
Explanation: NumPy arrays
End of explanation
z1 = zarr.array(a1, chunks=chunks, compression='blosc', 
                compression_opts=dict(cname='lz4', clevel=1, shuffle=2))
z1
z2 = zarr.empty_like(z1)
z2
profile_dask_copy(z1, z2, chunks, lock=True, dt=.02)
profile_dask_copy(z1, z2, chunks, lock=False, dt=0.02)
# for comparison, using blosc internal threads
%timeit -n3 -r5 z2[:] = z1
%prun z2[:] = z1
Explanation: Zarr arrays (in-memory)
End of explanation
h5f = h5fmem()
h5f
h1 = h5f.create_dataset('h1', data=a1, chunks=chunks, compression='lzf', shuffle=True)
h5d_diagnostics(h1)
h2 = h5f.create_dataset('h2', shape=h1.shape, chunks=h1.chunks, 
                        compression=h1.compression, compression_opts=h1.compression_opts, 
                        shuffle=h1.shuffle)
h5d_diagnostics(h2)
profile_dask_copy(h1, h2, chunks, lock=True, dt=0.1)
profile_dask_copy(h1, h2, chunks, lock=False, dt=0.1)
Explanation: Without the dask lock, we get better CPU utilisation. 
HDF5 datasets (in-memory)
End of explanation
c1 = bcolz.carray(a1, chunklen=chunks[0],
                  cparams=bcolz.cparams(cname='lz4', clevel=1, shuffle=2))
c1
c2 = bcolz.zeros(a1.shape, chunklen=chunks[0], dtype=a1.dtype, 
                 cparams=bcolz.cparams(cname='lz4', clevel=1, shuffle=2))
c2
profile_dask_copy(c1, c2, chunks, lock=True, dt=0.05)
# not sure it's safe to use bcolz without a lock, but what the heck...
profile_dask_copy(c1, c2, chunks, lock=False, dt=0.05)
# for comparison
%timeit -n3 -r5 c2[:] = c1
# for comparison
%timeit -n3 -r5 c1.copy()
Explanation: Bcolz carrays (in-memory)
End of explanation
<END_TASK>