m

    Marwan Sarieddine

    2 years ago
    Hi again everyone, I am trying to replicate the [Dask Cluster on Kubernetes deployment recipe](https://docs.prefect.io/orchestration/recipes/k8s_dask.html) and I face the following error
    AttributeError: 'FunctionTask' object has no attribute 'result'
    Wondering if you have encountered this before - One thing to note is if the flow is empty then it runs successfully … I am using prefect version
    0.10.7
    19 May 2020,07:027 	agent	INFO	Submitted for execution: Job prefect-job-d3fc6dc5
    19 May 2020,07:02:12 	prefect.CloudFlowRunner	INFO	Beginning Flow run for 'Static Dask Cluster'
    19 May 2020,07:02:12 	prefect.CloudFlowRunner	INFO	Starting flow run.
    19 May 2020,07:02:12 	prefect.CloudFlowRunner	DEBUG	Flow 'Static Dask Cluster': Handling state change from Scheduled to Running
    19 May 2020,07:02:12 	prefect.CloudFlowRunner	ERROR	Unexpected error: AttributeError("'FunctionTask' object has no attribute 'result'")
    Traceback (most recent call last):
      File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
        new_state = method(self, state, *args, **kwargs)
      File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 465, in get_flow_run_state
        for t in final_tasks
      File "/usr/local/lib/python3.7/site-packages/prefect/engine/executors/dask.py", line 196, in wait
        return self.client.gather(futures)
      File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 1931, in gather
        asynchronous=asynchronous,
      File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 780, in sync
        self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
      File "/usr/local/lib/python3.7/site-packages/distributed/utils.py", line 347, in sync
        raise exc.with_traceback(tb)
      File "/usr/local/lib/python3.7/site-packages/distributed/utils.py", line 331, in f
        result[0] = yield future
      File "/usr/local/lib/python3.7/site-packages/tornado/gen.py", line 735, in run
        value = future.result()
      File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 1790, in _gather
        raise exception.with_traceback(traceback)
      File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 569, in run_task
        default_handler = task.result_handler or self.flow.result_handler
    AttributeError: 'FunctionTask' object has no attribute 'result'
    19 May 2020,07:02:12 	prefect.CloudFlowRunner	DEBUG	Flow 'Static Dask Cluster': Handling state change from Running to Failed
    Chris White

    Chris White

    2 years ago
    Hi @Marwan Sarieddine - this is a symptom of your dask workers running an older version of Prefect (pre 0.11.0)
    m

    Marwan Sarieddine

    2 years ago
    @Chris White - thank you for your response - yep just updated the images to use 0.11.1 and it runs fine now
    As a follow up - I guess ideally I would want to try the dask-kubernetes approach of dynamically provisioning worker pods opposed to the static dask cluster implementation I mention above - so I tried to replicate the example on [Dask Kubernetes Environment](https://docs.prefect.io/orchestration/execution/dask_k8s_environment.html#overview) and I am getting this
    ast.literal_eval
    error with the logs: I am using prefect version
    0.11.1
    $ kubectl logs prefect-dask-job-6490fd77-df9b-4c03-9e05-0e5f64ea86ee-9czqp 
    Traceback (most recent call last):
      File "<string>", line 1, in <module>
      File "/usr/local/lib/python3.7/site-packages/prefect/__init__.py", line 1, in <module>
        import prefect.utilities
      File "/usr/local/lib/python3.7/site-packages/prefect/utilities/__init__.py", line 1, in <module>
        import prefect.utilities.logging
      File "/usr/local/lib/python3.7/site-packages/prefect/utilities/logging.py", line 252, in <module>
        configure_extra_loggers()
      File "/usr/local/lib/python3.7/site-packages/prefect/utilities/logging.py", line 247, in configure_extra_loggers
        loggers = literal_eval(context.config.logging.get("extra_loggers", "[]"))
      File "/usr/local/lib/python3.7/ast.py", line 91, in literal_eval
        return _convert(node_or_string)
      File "/usr/local/lib/python3.7/ast.py", line 90, in _convert
        return _convert_signed_num(node)
      File "/usr/local/lib/python3.7/ast.py", line 63, in _convert_signed_num
        return _convert_num(node)
      File "/usr/local/lib/python3.7/ast.py", line 55, in _convert_num
        raise ValueError('malformed node or string: ' + repr(node))
    ValueError: malformed node or string: <BoxList: ['dask_kubernetes.core', 'distributed.deploy.adaptive', 'kubernetes']>
    not quite sure why that is the case … seems like there is an attempt to extend the prefect loggers that fails …
    Chris White

    Chris White

    2 years ago
    Apologies for that issue - if you upgrade to 0.11.2 it should be resolved
    m

    Marwan Sarieddine

    2 years ago
    @Chris White - yep upgrading to 0.11.2 solves the problem - thanks