@Marvin Hey Marvin, I have a flow with multiple tasks in it. Some of the tasks returned large dataframes which when logged can be quite annoying. Any idea on how to disable that at flow and task level?

To suppress the logging of large dataframes or any task output in Prefect 3.x, you can adjust the

log_prints

setting for tasks and flows. By default, Prefect logs print statements at the INFO level, but you can disable this behavior: 1. Task Level: Set

log_prints=False

when defining a task to prevent logging of print statements within that task. 2. Flow Level: Similarly, set

log_prints=False

at the flow level to stop logging for all tasks within the flow unless overridden at the task level. Here's a simple example:

python
from prefect import flow, task

@task(log_prints=False)
def process_large_dataframe():
    # Your task logic here
    print("This will not be logged.")

@flow(log_prints=False)
def my_flow():
    process_large_dataframe()

my_flow()

For more details on logging configurations, you can refer to the Prefect logging documentation.

@Marvin For precision, the annoying log comes from Prefect not the flow code. I do not want the output of the task to be logged.

Finished in state Completed(message=None, type=COMPLETED, result=ResultRecord(metadata=ResultRecordMetadata(storage_key='/root/.prefect/storage/f6e4a36c29a9a090a83186835f2fe5d4', expiration=None, serializer=PickleSerializer(type='pickle', picklelib='cloudpickle', picklelib_version=None), prefect_version='3.1.2', storage_block_id=None), result=some_large_dataframe))

To suppress the logging of task results in Prefect 3.x, you can adjust the result persistence settings. By default, Prefect logs task results, which might include large dataframes. Here's how you can manage this: 1. Disable Result Persistence: You can disable result persistence for specific tasks by setting

persist_result=False

in the task decorator. This prevents Prefect from storing and thus logging the task's result. 2. Global Configuration: You can also configure this behavior globally using the

PREFECT_TASKS_DEFAULT_PERSIST_RESULT

setting, though this might affect all tasks and flows. Here's an example of how to disable result persistence for a task:

python
from prefect import flow, task

@task(persist_result=False)
def process_large_dataframe():
    # Your task logic here
    return some_large_dataframe

@flow
def my_flow():
    process_large_dataframe()

my_flow()

For more details, you can refer to the Prefect documentation on result persistence. This should help you avoid those annoying logs. Life's full of little annoyances, isn't it?

@Marvin this does not solve the issue. the flow is decorated with persist_result=False; same for the tasks which depends on each other.

It seems that the logging of task results, even when

persist_result=False

, is still causing issues. The message you're seeing is likely due to Prefect's internal logging of task completion states, which includes the result by default. Unfortunately, the standard Prefect configuration doesn't provide a direct way to suppress these specific log messages entirely. However, you might consider the following workaround: 1. Custom Logging Configuration: You can create a custom logging configuration to filter out specific log messages. This involves setting up a custom logging handler that ignores logs containing specific messages or patterns. 2. Modify Log Level: Change the log level to a higher threshold (such as

WARNING

or

ERROR

) for the specific logger responsible for these messages, though this might suppress other important informational logs as well. Here's a rough idea of how you might set up a custom logging configuration:

python
import logging
from prefect import flow, task
from prefect.logging import configure_logging

# Custom logging filter
class SuppressResultFilter(logging.Filter):
    def filter(self, record):
        return "Finished in state" not in record.getMessage()

# Configure logging with custom filter
configure_logging()
logger = logging.getLogger("prefect")
logger.addFilter(SuppressResultFilter())

@task
def process_large_dataframe():
    # Task logic
    return some_large_dataframe

@flow
def my_flow():
    process_large_dataframe()

my_flow()

This approach is more complex and requires careful handling of logging to ensure you don't miss other critical logs. For more detailed logging configuration, you might refer to the Prefect documentation or explore community solutions for similar issues. If only life were simpler, but alas, here we are.