https://dagster.io/ logo
#ask-ai
Title
# ask-ai
m

Martin Laurent

02/26/2024, 4:22 PM
In an asset sensor, how can I get the metadata of materialized asset that was detected
s

Scout

02/26/2024, 4:22 PM
In an asset sensor, you can access the metadata of a materialized asset by querying the event log for the latest materialization event of the asset. The documentation provides an example of how to access metadata from a historical materialization of one asset from inside another asset or an op, and a similar approach can be used within an asset sensor. Here's a step-by-step guide based on the documentation: 1. Obtain the
instance
attribute from the sensor's context. This instance is a
DagsterInstance
object that can interact with the event log. 2. Use the
instance
to query for the latest materialization event of the asset using the
get_latest_materialization_event
method, passing in the
AssetKey
of the asset you're interested in. 3. Access the
asset_materialization
property of the event to get the materialization object. 4. Retrieve the metadata from the materialization object's
metadata
property. Here is a conceptual example based on the documentation:
Copy code
python from dagster import sensor, SensorEvaluationContext, DagsterInstance, AssetKey  @sensor def my_asset_sensor(context: SensorEvaluationContext):     instance = context.instance  # Get the DagsterInstance from the context     asset_key = AssetKey(["my_asset"])  # Replace with your asset's key     materialization_event = instance.get_latest_materialization_event(asset_key)     if materialization_event:         materialization = materialization_event.asset_materialization         metadata = materialization.metadata  # Access the metadata dictionary         # Now you can use the metadata as needed
Please note that the actual code you write may vary depending on the specific details of your use case and the version of Dagster you are using. The example above is a conceptual guide and may require adjustments to fit your context.
m

Martin Laurent

02/26/2024, 4:26 PM
@Scout can I get the partition key of the asset
s

Scout

02/26/2024, 4:26 PM
Yes, you can get the partition key of a materialized asset from the materialization event. When you query the event log for the latest materialization event of an asset, the event object contains information about the partition key if the asset is partitioned. Here's how you can access the partition key based on the documentation: 1. Obtain the
instance
attribute from the sensor's context, which is a
DagsterInstance
object. 2. Use the
instance
to query for the latest materialization event of the asset using the
get_latest_materialization_event
method, passing in the
AssetKey
of the asset you're interested in. 3. Access the
event_log_entry
property of the event to get the event log entry object. 4. Retrieve the partition key from the
partition
property of the event log entry object. Here is a conceptual example:
Copy code
python from dagster import sensor, SensorEvaluationContext, DagsterInstance, AssetKey  @sensor def my_asset_sensor(context: SensorEvaluationContext):     instance = context.instance  # Get the DagsterInstance from the context     asset_key = AssetKey(["my_asset"])  # Replace with your asset's key     materialization_event = instance.get_latest_materialization_event(asset_key)     if materialization_event:         partition_key = materialization_event.event_log_entry.partition  # Access the partition key         # Now you can use the partition_key as needed
This example assumes that the asset is partitioned and that a materialization event for the asset exists. If the asset is not partitioned or if there is no materialization event, the
partition
property may be
None
or the event may not exist. Adjust the code to handle these cases as needed for your application.
m

Martin Laurent

02/26/2024, 4:30 PM
@Scout what are the available parameters in a asset_sensor
s

Scout

02/26/2024, 4:30 PM
In Dagster, when you define an
asset_sensor
, you can use several parameters to configure its behavior. The available parameters for an
asset_sensor
decorator are as follows: -
asset_key
(required): The
AssetKey
that the sensor monitors for materialization events. -
name
(optional): The name of the sensor. If not provided, it defaults to the name of the decorated function. -
job_name
(optional): The name of the job to be triggered by the sensor. This parameter is deprecated in favor of the
job
parameter. -
job
(optional): The job that the sensor will trigger when it fires. This replaces the
job_name
parameter. -
jobs
(optional): A list of jobs that the sensor will trigger when it fires. This is useful if you want the sensor to be able to trigger multiple jobs. -
minimum_interval_seconds
(optional): The minimum number of seconds that will elapse between sensor evaluations. This helps prevent sensors from running too frequently. -
description
(optional): A human-readable description of the sensor. -
default_status
(optional): The default status of the sensor (
RUNNING
or
STOPPED
). It determines whether the sensor starts as active or inactive. -
required_resource_keys
(optional): A set of resource keys that the sensor requires to run. This ensures that the necessary resources are available when the sensor executes. Here's an example of defining an
asset_sensor
with some of these parameters:
Copy code
python from dagster import asset_sensor, AssetKey, RunRequest, SensorEvaluationContext  @asset_sensor(     asset_key=AssetKey("my_asset"),     name="my_asset_sensor",     minimum_interval_seconds=30,     description="Triggers when my_asset is materialized." ) def my_asset_sensor(context: SensorEvaluationContext, asset_event):     # Sensor logic to create a RunRequest or SkipReason     return RunRequest(run_key="unique_run_key", run_config={})
Please note that the parameters and their usage might evolve over time, so it's always a good idea to consult the latest Dagster documentation for the most up-to-date information.
3 Views