dagster dev or the Dagster UI would display an error when loading jobs created with op or asset selections.@multi_asset s are in use has been drastically improved.the_asset.with_attributes is called without providing a group name. Previously, the existing group name was erroneously dropped. Thanks @ion-elgreco!dbt source freshness.MetadataValue in Essentials course. Thanks @stianthaulow!tag_keys argument on DagsterInstance.get_run_tagsis no longer optional. This has been done to remove an easy way of accidentally executing an extremely expensive database operation.AssetSelection.dagster-snowflake now requires snowflake-connector-python>=3.4.0@sling_assets accepts an optional name parameter for the underlying opdagster-openai library is now available.DagsterDbtTranslatorSettings called enable_duplicate_source_asset_keys that allows users to set duplicate asset keys for their dbt sources. Thanks @hello-world-bfree!ConfigurableResource.SlingConnectionResource where a blank keyword argument would be emitted as an environment variabledbt source freshness would cause an error.DbtArtifacts has been added for managing the behavior of rebuilding the manifest during development but expecting a pre-built one in production.send sample alert button now exists on both the alert policies page and in the alert policies editor to make it easier to debug and configure alerts without having to wait for an event to kick them off.SlingConnectionResource that raised an error when connecting to a database.graph_multi_assets with check_specs now support subsetting.run_retries.retry_on_op_or_asset_failures setting that can be set to false to make run retries only occur when there is an unexpected failure that crashes the run, allowing run-level retries to co-exist more naturally with op or asset retries. See the docs for more information.dagster dev now sets the environment variable DAGSTER_IS_DEV_CLI allowing subprocesses to know that they were launched in a development context.@multi_observable_source_asset decorator enables defining a set of assets that can be observed together with the same function.@sling_assets and Resource SlingConnectionResource have been added for the [dagster-embedded-elt.sling](http://dagster-embedded-elt.sling) package. Deprecated build_sling_asset, SlingSourceConnection and SlingTargetConnection.QueuedRunCoordinator.dagster-polars has been added as an integration. Thanks @danielgafni!@dbt_assets now supports loading projects with semantic models.@dbt_assets now supports loading projects with model versions.get_asset_key_for_model now supports retrieving asset keys for seeds and snapshots. Thanks @aksestok!@asset decorator causing a false positive in Pyright strict mode. Thanks @tylershunt!security_context field on the k8s_job_executor didn't correctly set the security context on the launched step pods. Thanks @krgn!ObserveResults with no data_version.FreshnessPolicys on observable source assets. These assets will be considered “Overdue” when the latest value for the “dagster/data_time” metadata value is older than what’s allowed by the freshness policy.DbtCliResource now include metadata from the dbt adapter response. This includes fields like rows_affected, query_id from the Snowflake adapter, or bytes_processed from the BigQuery adapter.k8s_job_executor.context.log.exception in a sensor or schedule did not properly capture exception information.@observable_source_asset-decorated functions can now return an ObserveResult. This allows including metadata on the observation, in addition to a data version. This is currently only supported for non-partitioned assets.AutoMaterializeRule.skip_on_not_all_parents_updated_since_cron class allows you to construct AutoMaterializePolicys which wait for all parents to be updated after the latest tick of a given cron schedule.AssetsDefinition.from_op. Dagster will infer asset inputs and outputs from the ins/outs defined on the @op in the same way as @graphs.default_executor_def argument. The default executor definition will be used for all op/asset jobs that don’t explicitly define their own executor.JobDefinition.run_request_for_partition now accepts a tags argument (Thanks @jburnich!)@multi_asset now accepts a resource_defs argument. The provided resources can be either used on the context, or satisfy the io manager requirements of the outs on the asset.define_asset_job and to_job now can now accept a partitions_def argument and a config argument at the same time, as long as the value for the config argument is a hardcoded config dictionary (not a PartitionedConfig or ConfigMapping)k8s_job_executor and docker_executor would sometimes return the same event lines twice in the command-line output for the step.@op decorator (Thanks Milos Tomic!)UnresolvedAssetJobDefinition now supports the run_request_for_partition method.AssetOut and AssetInSoftware-defined assets are now marked fully stable and are ready for prime time - we recommend using them whenever your goal using Dagster is to build and maintain data assets.
You can now organize software defined assets into groups by providing a group_name on your asset definition. These assets will be grouped together in Dagit.
Software-defined assets now accept configuration, similar to ops. E.g.
from dagster import asset @asset(config_schema={"iterations": int}) def my_asset(context): for i in range(context.op_config["iterations"]): ...
Asset definitions can now be created from graphs via AssetsDefinition.from_graph:
@graph(out={"asset_one": GraphOut(), "asset_two": GraphOut()}) def my_graph(input_asset): ... graph_asset = AssetsDefinition.from_graph(my_graph)
execute_in_process and GraphDefinition.to_job now both accept an input_values argument, so you can pass arbitrary Python objects to the root inputs of your graphs and jobs.
Ops that return Outputs and DynamicOutputs now work well with Python type annotations. You no longer need to sacrifice static type checking just because you want to include metadata on an output. E.g.
from dagster import Output, op @op def my_op() -> Output[int]: return Output(5, metadata={"a": "b"})
You can now automatically re-execute runs from failure. This is analogous to op-level retries, except at the job level.
You can now supply arbitrary structured metadata on jobs, which will be displayed in Dagit.
The partitions and backfills pages in Dagit have been redesigned to be faster and show the status of all partitions, instead of just the last 30 or so.
The left navigation pane in Dagit is now grouped by repository, which makes it easier to work with when you have large numbers of jobs, especially when jobs in different repositories have the same name.
The Asset Details page for a software-defined asset now includes a Lineage tab, which makes it easy to see all the assets that are upstream or downstream of an asset.
This release marks the official transition of software-defined assets from experimental to stable. We made some final changes to incorporate feedback and make the APIs as consistent as possible:
fs_asset_io_manager in favor of merging its functionality with fs_io_manager. fs_io_manager is now the default IO manager for asset jobs, and will store asset outputs in a directory named with the asset key. Similarly, removed adls2_pickle_asset_io_manager, gcs_pickle_asset_io_manager , and s3_pickle_asset_io_manager. Instead, adls2_pickle_io_manager, gcs_pickle_io_manager, and s3_pickle_io_manager now support software-defined assets.@asset decorator and AssetIn has been deprecated. Users should use key_prefix instead.define_asset_job (replacing AssetGroup.build_job), and arbitrary sets of assets can be materialized using the standalone function materialize (replacing AssetGroup.materialize).outs property of the previously-experimental @multi_asset decorator now prefers a dictionary whose values are AssetOut objects instead of a dictionary whose values are Out objects. The latter still works, but is deprecated.OpExecutionContext called output_asset_partition_key is now deprecated in favor of asset_partition_key_for_outputget_event_records method on DagsterInstance now requires a non-None argument event_records_filter. Passing a None value for the event_records_filter argument will now raise an exception where previously it generated a deprecation warning.events_for_asset_key and get_asset_events, which have been deprecated since 0.12.0.dbt_assets = load_assets_from_dbt_project(..., node_info_to_asset_key=lambda node_info: AssetKey(node_info["name"]).prior_attempts_count parameter is now removed from step-launching APIs. This parameter was not being used, as the information it held was stored elsewhere in all cases. It can safely be removed from invocations without changing behavior.FileCache class has been removed.A new define_asset_job function allows you to define a selection of assets that should be executed together. The selection can be a simple string, or an AssetSelection object. This selection will be resolved into a set of assets once placed on the repository.
from dagster import repository, define_asset_job, AssetSelection string_selection_job = define_asset_job( name="foo_job", selection="*foo" ) object_selection_job = define_asset_job( name="bar_job", selection=AssetSelection.groups("some_group") ) @repository def my_repo(): return [ *my_list_of_assets, string_selection_job, object_selection_job, ]
[dagster-dbt] Assets loaded with load_assets_from_dbt_project and load_assets_from_dbt_manifest will now be sorted into groups based on the subdirectory of the project that each model resides in.
@asset and @multi_asset are no longer considered experimental.
Adds new utility methods load_assets_from_modules, assets_from_current_module, assets_from_package_module, and assets_from_package_name to fetch and return a list of assets from within the specified python modules.
Resources and io managers can now be provided directly on assets and source assets.
from dagster import asset, SourceAsset, resource, io_manager @resource def foo_resource(): pass @asset(resource_defs={"foo": foo_resource}) def the_resource(context): foo = context.resources.foo @io_manager def the_manager(): ... @asset(io_manager_def=the_manager) def the_asset(): ...
Note that assets provided to a job must not have conflicting resource for the same key. For a given job, all resource definitions must match by reference equality for a given key.
A materialize_to_memory method which will load the materializations of a provided list of assets into memory:
from dagster import asset, materialize_to_memory @asset def the_asset(): return 5 result = materialize_to_memory([the_asset]) output = result.output_for_node("the_asset")
A with_resources method, which allows resources to be added to multiple assets / source assets at once:
from dagster import asset, with_resources, resource @asset(required_resource_keys={"foo"}) def requires_foo(context): ... @asset(required_resource_keys={"foo"}) def also_requires_foo(context): ... @resource def foo_resource(): ... requires_foo, also_requires_foo = with_resources( [requires_foo, also_requires_foo], {"foo": foo_resource}, )
You can now include asset definitions directly on repositories. A default_executor_def property has been added to the repository, which will be used on any materializations of assets provided directly to the repository.
from dagster import asset, repository, multiprocess_executor @asset def my_asset(): ... @repository(default_executor_def=multiprocess_executor) def repo(): return [my_asset]
The run_storage, event_log_storage, and schedule_storage configuration sections of the dagster.yaml can now be replaced by a unified storage configuration section. This should avoid duplicate configuration blocks with your dagster.yaml. For example, instead of:
# dagster.yaml run_storage: module: dagster_postgres.run_storage class: PostgresRunStorage config: postgres_url: { PG_DB_CONN_STRING } event_log_storage: module: dagster_postgres.event_log class: PostgresEventLogStorage config: postgres_url: { PG_DB_CONN_STRING } schedule_storage: module: dagster_postgres.schedule_storage class: PostgresScheduleStorage config: postgres_url: { PG_DB_CONN_STRING }
You can now write:
storage: postgres: postgres_url: { PG_DB_CONN_STRING }
All assets where a group_name is not provided are now part of a group called default.
The group_name parameter value for @asset is now restricted to only allow letters, numbers and underscore.
You can now set policies to automatically retry Job runs. This is analogous to op-level retries, except at the job level. By default the retries pick up from failure, meaning only failed ops and their dependents are executed.
[dagit] The new repository-grouped left navigation is fully launched, and is no longer behind a feature flag.
[dagit] The left navigation can now be collapsed even when the viewport window is wide. Previously, the navigation was collapsible only for small viewports, but kept in a fixed, visible state for wide viewports. This visible/collapsed state for wide viewports is now tracked in localStorage, so your preference will persist across sessions.
[dagit] Queued runs can now be terminated from the Run page.
[dagit] The log filter on a Run page now shows counts for each filter type, and the filters have higher contrast and a switch to indicate when they are on or off.
[dagit] The partitions and backfill pages have been redesigned to focus on easily viewing the last run state by partition. These redesigned pages were previously gated behind a feature flag — they are now loaded by default.
[dagster-k8s] Overriding labels in the K8sRunLauncher will now apply to both the Kubernetes job and the Kubernetes pod created for each run, instead of just the Kubernetes pod.
docker_executor to run each step of a job in a different Docker container.env_vars field to the EcsRunLauncher that allows you to configure environment variables in the ECS task for launched runs.env_vars field on K8sRunLauncher and k8s_job_executor can now except input of the form ENV_VAR_NAME=ENV_VAR_VALUE, and will set the value of ENV_VAR_NAME to ENV_VAR_VALUE. Previously, it only accepted input of the form ENV_VAR_NAME, and the environment variable had to be available in the pod launching the job.IOManager.load_input was called to load an asset that was not being materialized as part of the run, the provided context would not include the metadata for that asset. context.upstream_output.metadata now correctly returns the metadata on the upstream asset.list[str]) as the type of an input would raise an exception.metadata parameter) and viewed in dagit. You can use it to track code owners, link to docs, or add other useful information.config_schema. If you attempt to materialize an asset with a config schema in Dagit, you'll be able to enter the required config via a modal.k8s_job_executor where ops with long names sometimes failed to create a pod due to a validation error with the label names automatically generated by Dagster.fs_asset_io_manager has been removed in favor of merging its functionality with fs_io_manager. fs_io_manager is now the default IO manager for asset jobs, and will store asset outputs in a directory named with the asset key.k8s_job_executor’s max_conccurent configuration. Thanks @fahadkh!fs_io_manager to incorrectly handle assets associated with upstream assets. Thanks @aroig!protobuf version 3 due to a backwards incompatible change in the probobuf version 4 release.AssetGroup.build_job(), you can define selections which select subsets of the loaded dbt project.load_assets_from_dbt_manifest function now supports an experimental select parameter. This allows you to use dbt selection syntax to select from an existing manifest.json file, rather than having Dagster re-compile the project on demand.OpExecutionContext now exposes an asset_key_for_output method, which returns the asset key that one of the op’s outputs corresponds too.python -m dagster.daemon.non_argument_deps parameter for the asset and multi_asset decorators can now be a set of strings in addition to a set of AssetKey.InvalidSubsetError. This is now fixed.AssetKeys.snowflake_io_manager would sometimes raise an error with pandas 1.4 or later installed.DagsterExecutionStepNotFoundError. This is now fixed.