Locking in dask.multiprocessing.get and adding metadata to HDF
<p>Performing an ETL-task in pure Python, I would like to collect error metrics as well as metadata for each of the raw input files considered (error metrics are computed from error codes provided in the data section of the files while metadata is stored in headers). Here's pseudo-code for the whole procedure:</p>

<pre><code>import pandas as pd
import dask
from dask import delayed
from dask import dataframe as dd

META_DATA = {}  # shared resource
ERRORS = {}  # shared resource

def read_file(file_name):
    global META_DATA, ERRORS

    # step 1: process headers
    headers = read_header(file_name)
    errors = {}
    data_bfr = []

    # step 2: process data section
    for line in data_section:
        content_id, data = parse_line(line)
        if contains_errors(data):
            errors[content_id] = get_error_code(data)
        else:
            data_bfr.append(content_id, data)

    # ---- Part relevant for question 1 ----
    # step 3: acquire lock for shared resource and write metadata
    with lock.acquire():
        write_metadata(file_name, headers)  # stores metadata in META_DATA[file_name]
        write_errors(file_name, errors)  # stores error metrics in ERRORS[file_name]

    return pd.DataFrame(data=data_bfr,...)

with set_options(get=dask.multiprocessing.get):
    df = dd.from_delayed([delayed(read_file)(file_name) \
                          for file_name in os.listdir(wd)])

    # ---- Part relevant for question 2 ----
    df.to_hdf('data.hdf', '/data', 'w', complevel=9, \
        complib='blosc',..., metadata=(META_DATA, ERRORS))
</code></pre>

<p>For each input file <code>read_file</code> returns a <code>pd.DataFrame</code>, further writing relevant metadata and error metrics to shared resources. I am using <code>dask</code>'s multiprocessing scheduler to compute a <code>dask.dataframe</code> from a list of delayed <code>read_file</code>-operations. </p>

<ul>
<li><strong>Question 1</strong>: Each of the <code>read_file</code>-operations writes to the shared resources <code>META_DATA</code> and <code>ERRORS</code>. What do I have to do to implement a proper locking strategy that works with <code>dask.multiprocessing.get</code>? Would it be sufficient to write the metadata and error information to the collections from within a <code>with locket.lock_file('.lock'):</code>-context? Does <code>multiprocessing.RLock</code> work? What do I have to do to initialize the lock to work with <code>dask</code>? More fundamentally, how can I declare <code>META_DATA</code> and <code>ERRORS</code> as shared resources in <code>dask</code>?</li>
<li><strong>Question 2</strong>: If possible, I would like to annotate HDF-data with metadata and error metrics. From a <a href="http://stackoverflow.com/questions/35013926/collecting-attributes-from-dask-dataframe-providers">question on "Collecting attributes from dask dataframe providers"</a>, I learned that <code>dask</code> currently does not support adding metadata to dataframes, but is it possible for the information be written to HDF? If so, how to handle the access to the shared resources in this case?</li>
</ul>

<h3>Don't depend on Globals</h3>

<p>Dask works best with <a href="http://toolz.readthedocs.io/en/latest/purity.html#function-purity" rel="nofollow">pure functions</a>.  </p>

<p>In particular, your case is a limitation in Python, in that it (correctly) doesn't share global data between processes.  Instead, I recommend that you return data explicitly from functions:</p>

<pre><code>def read_file(file_name):
    ...
    return df, metadata, errors

values = [delayed(read_file)(fn) for fn in filenames]
dfs      = [v[0] for v in values]
metadata = [v[1] for v in values]
errors   = [v[2] for v in values]

df = dd.from_delayed(dfs)

import toolz
metadata = delayed(toolz.merge)(metadata)
errors = delayed(toolz.merge)(errors)
</code></pre>

python pandas locking metadata dask