Best practice to generate more dynamic downstream partitions from one asset

[ArcticXWolf]

Hello everyone,

we have a usecase where we get xml files delivered from customers and need to transform them into individual reports (per FILE, not per customer). The xml files get delivered as a zip file and we need to be able to reprocess such a file from time to time (for example for a backfill).
I tried for a long time, but it seems like dagster does not support this kind of usecase very well or I cannot figure it out from the documentation.

My idea was the following assets:

1. xml_collection asset which is partitioned with the zipfile name and it is an AssetSpec because it materializes externally
2. xml asset which is each single xml file (partitioned two-way with the zipfile name and with the xml file name)
3. xml_report asset which is the report corresponding to a single xml file (partitioned the same as the xml asset)

The xml_collection should be materialized by a sensor (that tracks the ingestion spot (SFTP)), but the others should be materialized via Declarative Automation (so they should automaterialize as soon as a collection arrives).

I was able to create the xml_collection AssetSpec, its corresponding partition and a sensor that creates an asset materialization for the xml_collection.

However now I want to write the materialization of the xml assets from that collection and I do not know any way of creating multiple asset materializations from one single asset.

Is dagster even the correct tool for this usecase? I've seen the discussion at #28678 and at #9559 but they have no solution that solves my setup.

---

Just some better example in code:

import dagster as dg
xml_collection_partition = dg.DynamicPartitionsDefinition(
    name="xml_collection_filename"
)

xml_collection = dg.AssetSpec(
    key="xml_collection", partitions_def=xml_collection_partition
)

@dg.sensor(minimum_interval_seconds=30)
def xml_collection_sensor(
    context: dg.SensorEvaluationContext,
) -> dg.SensorResult:
    zips_on_sftp = ... # get from sftp
    asset_materializations = []
    partition_requests = []

    # do some stuff to only process new and unseen zipfiles
    ...

    for zip_file_name in zips_on_sftp:
        asset_materializations.append(dg.AssetMaterialization(
            asset_key=xml_collection.key,
            partition=zip_file_name,
        ))
        partition_requests.append(
            xml_collection_partition.build_add_request(
                [zip_file_name]
            )
        )

    return dg.SensorResult(
        asset_events=asset_materializations,
        dynamic_partitions_requests=partition_requests,
    )

With that above, I know need to do something like:

xml_filename_partition = dg.DynamicPartitionsDefinition(name="xml_filename")
xml_partition = dg.MultiPartitionsDefinition(
    {"collection": xml_collection_partition, "xml_filename": xml_filename_partition}
)

@dg.asset(deps=[xml_collection], partitions_def=xml_partition, automation_condition=dg.AutomationCondition.eager())
def xml(context: dg.AssetExecutionContext): ... 

@dg.asset(deps=[xml], partitions_def=xml_partition, automation_condition=dg.AutomationCondition.eager())
def xml_report(context: dg.AssetExecutionContext): 
    # do some calculation for the report
    

But I dont know how to fill the materializations there (since I need to create MULTIPLE xml asset materializations from ONE xml_collection materialization).
Also the reason why we dont combine all xmls per delivery into one xmls asset is because every single one of them is allowed to fail and not continue. A different example that shows this well would be getting a delivery of PDFs and having to do OCR on them.

[ArcticXWolf]

I've found a partial solution. We can add to the xml_partition in the materialization of the xml_collection asset and this will trigger the correct xml materializations. However we now run into this issue of the MultiPartition being full cross product: #13139 (comment)

So I dont see how this usecase is currently feasible in dagster :(