import dask.array as da
import numpy as np
import sparse

def corr_on_chunked(chunk1, chunk2, corr_thresh=0.9):
    return sparse.COO.from_numpy((np.dot(chunk1, chunk2.T) > corr_thresh))

x = da.random.random((10000, 1000), chunks=(1000, 1000))

z = da.core.atop(corr_on_chunked, 'ik', x, 'ij', x.T, 'jk', dtype=x.dtype,
                 concatenate=True)
z = da.triu(z, k=1)

# z.visualize('dask.png', optimize_graph=True, rankdir='LR')
# z = z.compute()