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!