https://dagster.io/ logo
#ask-community
Title
# ask-community
z

Zeeshan Nehal

07/25/2023, 4:00 AM
Hi all, I am seeking some help with
DynamicPartitionsDefinition
for assets, in my case, I am using
MultiPartitionsDefinition
which consist of
DynamicPartitionsDefinition
which pull a list of ids from db on run-time, and another
DailyPartitionsDefinition
to execute asset daily based on date partition. It is strange using the dagester's sensor that I am able to see dynamic partition executing new ids being pulled from DB, but not removing the previous ids partition that was deleted. If I try to run manually using UI to materialize it gives the list of all historical dyanmic partitioned values (ids). In the second part I noticed that I cannot use freshness policy for ``DynamicPartitionsDefinition\, MultiPartitionsDefinition`
s

sandy

07/25/2023, 3:52 PM
which consist of
DynamicPartitionsDefinition
which pull a list of ids from db on run-time
when you say "on run-time", what exactly do you mean? and how are you deleting the previous ids?
z

Zeeshan Nehal

07/26/2023, 12:45 AM
Copy code
get_fp_ads_def = DynamicPartitionsDefinition(name="get_fp_ad_ids")
date_partition_def = HourlyPartitionsDefinition(start_date=datetime(2023, 7, 23), timezone="UTC")


@asset(name="asset_zee-test-partitions_multi_asset",
       partitions_def=MultiPartitionsDefinition(
           {
               'ad': get_fp_ads_def,
               'date': date_partition_def

           }
       ),
       auto_materialize_policy=AutoMaterializePolicy.eager()
       )
only flagged ids are being pulled from DB for this job, so it always changes. I used dynamic partition for this case, also how I can use date partition per dynamic items like ids list from DB is ['a', 'b', 'c'] having specific start and end date.
@sandy how to set the limit of concurrency of task for sensor ? In my case I have
MultiPartitionsDefinition
due to
DynamicPartitionsDefinition
in it, I want to set the max concurrent limit to 2 partition jobs concurrently?
@sandy I did get some hacky ways to get around multipartition in sensor, please provide feedback on it combining the partition key of
Dynamicpartitiondefinition
and `dailypartitionsdefinition`:
Copy code
@sensor .... date_for_partition = date_partition_def.start.strftime('%Y-%m-%d')
  
    <http://context.log.info|context.log.info>("new partition key: {}".format(f'xxyyxx|{date_for_partition}'))
    return SensorResult(
        # this support two partition keys else we need to use Multipartition tags,then the  dynamic request will be an issue
        run_requests=[RunRequest(partition_key=f'{ad_name}|{date_for_partition}') for ad_id in new_campaigns],
        dynamic_partitions_requests=[get_fp_ads_def.build_add_request(new_campaigns),
                                     get_fp_ads_def.build_delete_request(delete_campaigns)]
    )
2 Views