How do I clean up older data from the database?

[alangenfeld]

After using dagster for an extended period of time, my database storage has grown too large. Can I remove older content to reduce the size?

[alangenfeld]

One option is to use the python APIs against the DagsterInstance to query for older runs and delete them. This is a destructive operation that will remove the events, tags, and run record from the database. This will remove dagster s understanding that this run ever occurred, which can be particularly impactful to partitioned jobs and assets.

Perform this operation with great care.

An example script would look something like this.

import datetime

from dagster import DagsterInstance, RunsFilter

instance = DagsterInstance.get()

# define the time threshold for what is old enough, this example uses 1 week
week_ago = datetime.datetime.now() - datetime.timedelta(days=7)

old_run_records = instance.get_run_records(
    filters=RunsFilter(created_before=week_ago),
    limit=10,  # limit how many are fetched at a time, perform this operation in batches
    ascending=True,  # start from the oldest
)

# in this simple example we delete serially
# for higher throughput you could parallelize with threads
for record in old_run_records:
    # delete all the database contents for this run
    instance.delete_run(record.dagster_run.run_id)

[alangenfeld]

No this won't effect the sensor/schedule ticks or cursors which can be managed with these data retention setttngs
https://docs.dagster.io/deployment/dagster-instance#data-retention

[Lukashab]

@alangenfeld do you know if this should work even if I am using helm chart to deploy dagster with several code locations? Right now I have dedicated code location with one job (with schedule) that should cleanup the old records, but as it is in a separate code location, it seems that the dagster instance "does not see" the event logs of other code locations and it always removes 0 records from the table.

I have a following code:

@op
def delete_old_event_logs(metrics_client: MetricsClient) -> None:
    """
    Deletes event logs from storage that are older than EVENT_LOG_ROTATION_DAYS
    """
    instance = DagsterInstance.get()
    date_from = utcnow() - timedelta(days=settings.EVENT_LOG_ROTATION_DAYS)
    batch_size = 100

    runs_deleted = 0
    deleted_metric = EventLogDeletedGauge(
        prometheus=metrics_client.prometheus
    ).metric.labels(environment=settings.ENV_FOR_DYNACONF)

    while True:
        old_run_records = instance.get_run_records(
            filters=RunsFilter(created_before=date_from),
            ascending=True,
            limit=batch_size,
        )

        if len(old_run_records) == 0:
            break

        for record in old_run_records:
            instance.delete_run(record.dagster_run.run_id)

        deleted_metric.inc(len(old_run_records))

    logger.info(f"Deleted {runs_deleted} runs")
    ```

[prha]

Responded in #18807

[NiallRees]

Is there a way to do this in bulk? We have around 10000 run entries per day, this is slow to execute. Is there an easy way to get a database connection to just run a command like:

DELETE FROM public.event_logs
WHERE timestamp < '2024-02-01';

I realise this isn't recommended but happy to take the risk.

[SiviP-Glossai]

Hi,

I am doing just that.. Happy to share my code with you.

I do have a problem where deleting doesn't really free up space. What did you end up doing?

BTW we are using RDS

[dmosesson]

Will this trigger the relevant cleanup to take place as well (like VACUUM/VACUUM FULL)? If not I think this might make things faster, but from a disk space perspective, posgres will hold on the space.

[alangenfeld]

Correct, you may need to take additional actions or tweak any automatic vacuum settings depending on the specific database and version in use.

[HuyNguyenNeyuData]

If I want to keep ASSET_MATERIALIZATION and ASSET_OBSERVATION event for UI, is there a way to keep them?

[prha]

There's not an API to retain whitelisted event types from runs while deleting others. You could manually delete certain records from your DB, but it might affect some views in Dagster (run stats, step stats, retries, asset history, etc).

That being said, if you are comfortable manually deleting records from the event log, it should always be safe to delete events from the event log that have a null value for dagster_event_type.

[AndreaGiardini]

That being said, if you are comfortable manually deleting records from the event log, it should always be safe to delete events from the event log that have a null value for dagster_event_type.

@prha Is that because those events belong to runs that are not there anymore or what? Just trying to understand how that could happen.

[prha]

No, it's because we have two types of logs: structured and unstructured.

Structured logs will always have a dagster_event_type populated, and are required for dagster to keep track of what has happened. They power things like asset history, step retries, run retries, etc.

Unstructured logs (dagster_event_type is null) are just strings that are output by the ops/assets using context.log and don't contribute to dagster's understanding of what has happened in the run (but may contain useful debugging info on a per-run basis). That info generally has diminishing value after the run has completed.

[cleboo]

Is there a clean way to also remove the associated folders on disk? I have a disk that's slowly getting filled up with the contents of intermediate results of runs.

[cleboo]

I hacked together this approach that works locally and in a docker setup with local storage only:

def do_run_cleanup(context: OpExecutionContext):
    # ... find run to delete -> run ...
    path = Path(context.instance._local_artifact_storage.storage_dir) / Path(
        run.dagster_run.run_id
    )
    if path.is_dir():
        shutil.rmtree(path)

But that is going to break for a more involved storage setup (kubernetes, s3, ...).

[tiberiuana]

Just for inspiration, we are currently cleaning up the disk by blindly deleting storage older than 30 days.

find storage/ -type f -mtime +30 -exec rm -f {} \;

No guarantee this is a good idea, I can only confirm this hasn't broken our production instance (yet).

[stefanadelbert]

I'm doing the same. I've built it into a dagster job which I run periodically and it works well.

[garethbrickman]

In Dagster+, runs can be deleted via an authorized call to GraphQL API using the runId with the deleteRun mutation. ­An Organization Admin would need to provision a user token from in the ​Organization settings​ > ​Tokens​ tab. Below is an example script:

import os
from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport

# Define the endpoint URL and token
org_name = "your-org-name"
base_url = f"https://{org_name}.dagster.cloud/"
deployment_name = "prod"  # string name for actual deployments, branch deployments use a string ID of their deploymentId
url = base_url + deployment_name + "/graphql"
token = os.getenv("DAGSTER_CLOUD_USER_TOKEN")  # a user token generated from the Organization Settings page in Dagster+. Note: use a user token, not agent token

# Define the transport with the endpoint URL and any headers if needed
transport = RequestsHTTPTransport(
    url=url,
    headers={
        "Dagster-Cloud-Api-Token": token,
    },
    use_json=True,
    timeout=60,
)

# Instantiate the client
client = Client(transport=transport)

# Define the GraphQL mutation
delete_run_mutation = gql("""
mutation DeleteRun($runId: String!) {
  deleteRun(runId: $runId) {
    __typename
    ... on DeletePipelineRunSuccess {
      runId
    }
    ... on RunNotFoundError {
      runId
    }
    ... on PythonError {
      message
      stack
    }
  }
}
""")

# Define the query variables
query_variables = {
    "runId": "your-run-id"  # replace with the actual run ID you want to delete
}

# Execute the mutation
try:
    result = client.execute(delete_run_mutation, variable_values=query_variables)
    print(result)
except Exception as e:
    print(f"An error occurred: {e}")

[stefanadelbert]

FWIW, this is also possible for the community edition. I can confirm that it does remove the run such that it disappears from the Dagster UI. I can also comfirm that LogMessageEvents available from GraphQL for a particular run were also no longer available after deleting the run via GraphQL.
So, my assumption is that it is cleaning up database entries associated with the run, including logs.
I will consider implementing a cleanup job in Dagster which fetches all runs older than some threshold and then deletes those runs, either using GraphQL or the dagster instance.