https://prefect.io
Join Slack
Hello. Is it possible to stream task runner initialisation logs to the Cloud UI logs? I am using Das...
# ask-community
v
Hello. Is it possible to stream task runner initialisation logs to the Cloud UI logs? I am using DaskTaskRunner and user can't see logs about cluster creation and dashboard url.
And actually errors are very abstract. I can see more details only in pod logs.
c
Hi Vadym, do you have task loggers in your prefect tasks? DaskTaskRunners will / can collect logs into the Prefect UI, although it is slightly delayed just because of the nature of the processes having to spin up / tear down / log ship, and batch write to the cloud
but if you are logging inside the task with a task logger, then you should see logs in the UI, although they can be delayed
v
There is no logs from DaskTaskRunner even after task is completed.
c
Do you have a logger inside the task?
v
I am not sure I understand your question DaskTaskRunner starts before all tasks.
c
You can have a task logger, and a flow logger: @task(name=“log-example-task”) def logger_task(): logger = get_run_logger() logger.info(“INFO level log message from a task.“) @flow(name=“log-example-flow”) def logger_flow(): logger_task()
If you don’t have a task logger inside the task for Dask, logs are not being surfaced for that task
v
c
Gotcha, I completely misundertood
🙌 1
what does your task definition look like?
It looks like that should be adding info logs to the logger
before the connection is made
v
Copy code
@task
def write_data(storage: Storage, effective_date: datetime.date) -> None:
    df = pd.DataFrame({'col_1': ['1', '2', '3'], 'col_2': [1, 2, 3]})

    df['date'] = effective_date
    storage.save(dd.from_pandas(df, npartitions=1), SAMPLE_NAMESPACE)
in the task I don't have access to the cluster, I just submit it to the task_runner I specified in flow attributes
c
where is the dask cluster spun up from the task runner?
v
Does it matter I can pass any cluster_class
Copy code
DaskTaskRunner(
    cluster_class="dask_cloudprovider.aws.FargateCluster",
    cluster_kwargs={"n_workers": 4, "image": "my-prefect-image"},
)
or pass existing address
Copy code
@flow(task_runner=DaskTaskRunner(address="<http://my-dask-cluster>"))
def my_flow():
    ...
The main question - how to see logs from DaskTaskRunner in Cloud UI logs?
Because I see them on prod pod I am running the flow
c
That’s what I’m trying to suggest - when you call the task runner, you need to log the output of where it’s being called from
the taskrunner itself logs, but if you don’t have a logger defined in your flow where it is being instantiated
there is nothing surfacing it
Copy code
Logging in flows¶
To log from a flow, retrieve a logger instance with get_run_logger(), then call the standard Python logging methods.


from prefect import flow, get_run_logger

@flow(name="log-example-flow")
def logger_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("INFO level log message.")
Prefect automatically uses the flow run logger based on the flow context. If you run the above code, Prefect captures the following as a log event.


15:35:17.304 | INFO    | Flow run 'mottled-marten' - INFO level log message.
The default flow run log formatter uses the flow run name for log messages.
the DaskTaskRunner is surfacing .info logs, so if you are creating that in your flow, then I believe you would need a flow logger context to see those logs
v
but as I understand task runner starts before any code in the flow function
I see these logs in pod logs, but not in Cloud UI
4 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by