Dagster data migrations

[gnilrets]

I was wondering if anyone has attempted writing data migrations in dagster, similar to db migrations in Ruby on Rails or Django? The idea being that I have some data warehouse tables where I need to modify the columns or the table structure in non-trivial ways. I would only need to run this code once, but it might take hours to complete and certain new dagster processes shouldn't run until the migration is complete. I want to keep it in code so I have a history of it, but after it completes successfully, it should never be run again.

[EtienneT]

I was searching for the exact same thing and could not find anything. Simple things like renaming an asset while keeping the materialization events intact would be really cool. It feels like any serious data pipeline is going to evolve over time and dagster needs to be able to evolve with it.

Ideally I should be able to write migrations which dagster would run first before doing anything else.

Thinking about it though we have everything we need already to be able to do that before something more official is implemented. You could create a new job called migrations_job.

In your deployment script, before actually starting dagster, you would run the command line > dagster job execute migrations_job. This could run alembic migrations for example on your data, maybe do custom code to move stuff around in dagster like materialization etc. Then once the migrations have executed correctly, you would start dagster normally.

Another option could be to use dynamic partition in an asset called migrations. You would still run a job right before deployment, but then the job would only detect which migration needs to run, then create dynamic partitions in the migrations asset for each new migration to run and then trigger materializations in the migrations asset for those partitions. This way you could keep track of migrations as part of a dagster asset, keep logs etc.

[EtienneT]

Here is an example of code where you could do that:

from dagster import AssetExecutionContext, DynamicPartitionsDefinition, OpExecutionContext, asset, job, op

migrations_partitions_def = DynamicPartitionsDefinition(name="migrations")

@op
def migrations_op(context: OpExecutionContext):
    # Detect which migration # do we need to create as partition and add them to the dynamic partition definition.
    context.instance.add_dynamic_partitions('migrations', ['1', '2'])

@job
def migrations_job():
    migrations_op()

@asset(partitions_def=migrations_partitions_def)
def migrations(context: AssetExecutionContext):
    # Run a particular migration based on context.partition_key
    return True

[gnilrets]

That's an interesting idea. I haven't played much with partitions, but perhaps I'll try this out.