def run_dask_function(config):
    'Start a Dask Cluster using dask-kubernetes and run a function.\n\n    Talks to kubernetes to create `n` amount of new `pods` with a dask worker inside of each\n    forming a `dask` cluster. Then, a function specified from `config` is being imported and\n    run with the given arguments. The tasks created by this `function` are being run on the\n    `dask` cluster for distributed computation.\n\n    The config dict must contain the following sections:\n        * run\n        * dask_cluster\n        * output\n\n    Args:\n        config (dict):\n            Config dictionary.\n    '
    output_conf = config.get('output')
    if output_conf:
        path = output_conf.get('path')
        if (not path):
            raise ValueError('An output path must be provided when providing `output`.')
    cluster_spec = _generate_cluster_spec(config, kubernetes=False)
    cluster = KubeCluster.from_dict(cluster_spec)
    workers = config['dask_cluster'].get('workers')
    if (not workers):
        cluster.adapt()
    elif isinstance(workers, int):
        cluster.scale(workers)
    else:
        cluster.adapt(**workers)
    client = Client(cluster)
    client.get_versions(check=True)
    try:
        run = _import_function(config['run'])
        kwargs = config['run']['args']
        results = run(**kwargs)
    finally:
        client.close()
        cluster.close()
    if output_conf:
        bucket = output_conf.get('bucket')
        try:
            if bucket:
                aws_key = output_conf.get('key')
                aws_secret = output_conf.get('secret_key')
                _upload_to_s3(bucket, path, results, aws_key, aws_secret)
            else:
                os.makedirs(os.path.dirname(path), exist_ok=True)
                results.to_csv(path)
        except Exception:
            print('Error storing results. Falling back to console dump.')
            print(_df_to_csv_str(results))
    else:
        return results
Start a Dask Cluster using dask-kubernetes and run a function.

Talks to kubernetes to create `n` amount of new `pods` with a dask worker inside of each
forming a `dask` cluster. Then, a function specified from `config` is being imported and
run with the given arguments. The tasks created by this `function` are being run on the
`dask` cluster for distributed computation.

The config dict must contain the following sections:
    * run
    * dask_cluster
    * output

Args:
    config (dict):
        Config dictionary.
def run_dask_function(config):
    'Start a Dask Cluster using dask-kubernetes and run a function.\n\n    Talks to kubernetes to create `n` amount of new `pods` with a dask worker inside of each\n    forming a `dask` cluster. Then, a function specified from `config` is being imported and\n    run with the given arguments. The tasks created by this `function` are being run on the\n    `dask` cluster for distributed computation.\n\n    The config dict must contain the following sections:\n        * run\n        * dask_cluster\n        * output\n\n    Args:\n        config (dict):\n            Config dictionary.\n    '
    output_conf = config.get('output')
    if output_conf:
        path = output_conf.get('path')
        if (not path):
            raise ValueError('An output path must be provided when providing `output`.')
    cluster_spec = _generate_cluster_spec(config, kubernetes=False)
    cluster = KubeCluster.from_dict(cluster_spec)
    workers = config['dask_cluster'].get('workers')
    if (not workers):
        cluster.adapt()
    elif isinstance(workers, int):
        cluster.scale(workers)
    else:
        cluster.adapt(**workers)
    client = Client(cluster)
    client.get_versions(check=True)
    try:
        run = _import_function(config['run'])
        kwargs = config['run']['args']
        results = run(**kwargs)
    finally:
        client.close()
        cluster.close()
    if output_conf:
        bucket = output_conf.get('bucket')
        try:
            if bucket:
                aws_key = output_conf.get('key')
                aws_secret = output_conf.get('secret_key')
                _upload_to_s3(bucket, path, results, aws_key, aws_secret)
            else:
                os.makedirs(os.path.dirname(path), exist_ok=True)
                results.to_csv(path)
        except Exception:
            print('Error storing results. Falling back to console dump.')
            print(_df_to_csv_str(results))
    else:
        return results