Timezone for start and end times from GraphQL The response I get back from a GraphQL query like

{
  repositoriesOrError {
    ... on RepositoryConnection {
      nodes {
        jobs {
          runs {
            id
            startTime
            endTime
          }
        }
      }
    }
  }
}

looks like

{
  "data": {
    "repositoriesOrError": {
      "nodes": [
        {
          "jobs": [
            {
              "runs": [
                {
                  "id": "25718726-4265-45fb-b92f-ef656ec0703a",
                  "startTime": 1675048818.020318,
                  "endTime": 1675048823.704493
                }
              ]
            }
          ]
        }
      ]
    }
  }
}

The

startTime

and

endTime

have documentation like

startTime: Float

TYPE DETAILS

The

Float

scalar type represents signed double-precision fractional values as specified by IEEE 754.

What can we assume about the timezone or locale of those times?

is there a trick to override the partition of an output? --- Case I have: 1. At the end of the year (month =12) produce a file for each of the next 12 months 2. Every month the next year, refer to the file produced at the end of previous year, --- Pattern ideas: • Have a funtion that writes to file (step1) using specific file path logic to put it in the target folder for the next year • Then in step 2 can define a source asset with standard file path logic (io manager) to read the file If you can override the partition key of outputs, then it could be simply a factory pattern of assets x in range(12), setting the output partition = 2023-x-01, and use a standard iomanager. Maybe there are other solutions to this, e.g. partition mappings (but would need a custom mapping to map from current month to previous year end).

Am I over engineering things? Hi all. I have a sinking feeling that I’m building a rats nest of sensors and jobs to accomplish a fairly mundane task and now I need your help. Have I over engineered this? I have hundreds of thousands of CSV files in an S3 bucket. They have a specific prefix and I’m tasked with parsing them one by one (un-nesting an arbitrarily deep adjacent nodes hierarchy), sinking them to a data warehouse and finally giving them a new prefix to mark them as ‘archived’ or if the un-nesting fails ‘failed’. This last part is to signal to the sender that we’re done with the parsing. So I’ve built a sensor to list files under the original prefix, kicking of a job for each file to materialise an asset. 3 functions so far. Then I’ve created two *run_status_sensor*s to check if the run succeeded or failed. They both start the same job that configures an op to reset the prefix of the file from the original run, respectively. 7 functions in total, spread over assets, jobs, sensors and ops. Am I over engineering this? What would a simpler but still as robust solution look like?

It seems an IO-manager can be an input manager at the same time? But it seems the

@input_manager

decorator needs to decorate the function that will instantiate the internal io-manager?

(as long as i have defined

load_input

in the io_manager

Hello 👋 Is it necessary for the user code to have a

dagster.yaml

file when exposing it with gRPC ?

Hi, is there a way of running

snowflake_resource

on a SourceAsset? I'm trying to run a query on snowflake rather then using snowflake_pandas_io_manager as the dataset is too large to hold in local memory as a pandas dataframe

iris_harvest_data = SourceAsset(key="IRIS_DATASET", resource_defs=["snow_resource"])

not sure if i need to use resource_defs / how to config

Mapping[str, ResourceDefinition]

@asset(required_resource_keys={"snow_resource"})
def example_downstream(IRIS_DATASET):
    IRIS_DATASET.resources.snow_resource.execute_query("SELECT * LIMIT 150")


defs = Definitions(
    assets=[another_dataset, iris_harvest_data, example_downstream],
    resources={
        "snow_pd_io_manager": snowflake_pandas_io_manager.configured(
        {
            "account" : os.getenv("SNOWFLAKE_ACCOUNT"),
            "user": os.getenv("SNOWFLAKE_USER"),
            "password": os.getenv("SNOWFLAKE_PASSWORD"),
            "role" : "ACCOUNTADMIN",
            "warehouse": "COMPUTE_WH",
            "database": "CORE",
            "schema": "IAM"
        }),
        "snow_resource": snowflake_resource.configured(
        {
            "account" : os.getenv("SNOWFLAKE_ACCOUNT"),
            "user": os.getenv("SNOWFLAKE_USER"),
            "password": os.getenv("SNOWFLAKE_PASSWORD"),
            "role" : "ACCOUNTADMIN",
            "warehouse": "COMPUTE_WH",
            "database": "DAGSTER_TEST_ASSETS",
            "schema": "PUBLIC"
        })
    },
)

Quick question regarding loggers, as I'm missing something that is probably obvious. How can I provide a configuration to a logger defined like this

loggers_by_deployment_name = {
    "deploy": {"prod_logger": prod_logger},
    "local": {"console_logger": colored_console_logger},
}


env_config = dotenv_values(".env")

deployment_name = env_config["DAGSTER_DEPLOYMENT"]

active_logger = loggers_by_deployment_name[deployment_name]

defs = Definitions(assets=[some_asset], loggers=active_logger)

I need to configure my

prod_logger

. I have created a YAML config file, but I don't know how to provide it. I've read the documentation several time and I only saw the possibility to pass a config from the command line to a job, or from the dagit launchpad to a job.

hey team - logged an issue that we're seeing with the Dagster Cloud

hybrid_branch_deploy

action where it does not properly link to the branch deployment url.

Hello, I've upgraded to 1.1.18 and getting the following error in Dagit:

Operation name: ShouldShowNux

Message: [Errno 13] Permission denied: '/opt/dagster/dagster_home/.nux'

Path: ["shouldShowNux"]

Locations: [{"line":2,"column":3}]

Dagster is deployed using the Helm chart, and has the following mount:

/opt/dagster/dagster_home/dagster.yaml from dagster-instance (rw,path="dagster.yaml")

It also seems [it would not be displayed anyway as it is only shown on

localhost

](https://github.com/dagster-io/dagster/blob/master/js_modules/dagit/packages/app/src/NUX/CommunityNux.tsx#L27) -- but we get the GraphQL error anyway. Hence, indeed,

/opt/dagster/dagster_home/.nux

cannot be written. Is there a way to avoid this and disable the popup in the chart?

Question regarding an error I'm getting at the moment. If anyone has any insight that would be great. I have an Airbyte asset (one of many, but all are showing same behaviour) configured via:

airbyte_nasdaq_assets = build_airbyte_assets(
    connection_id=AIRBYTE_NASDAQ_CONNECTION_ID,
    destination_tables=["nasdaq_end_of_day_data"],
    asset_key_prefix=["timescale_source"],
)

and this (and two others) are feeding to a downstream dependency which is a dbt asset. So very similar to the modern data stack example project. When I use the UI to materialize the assets - either selectively or the full workflow - I get an error

2023-02-16 14:34:16 +0000 - dagster - DEBUG - __ASSET_JOB - 91cb0a21-d1d0-4a94-896f-da41410b8e52 - 12712 - airbyte_sync_b9e15 - ASSET_MATERIALIZATION - Materialized value timescale_source nasdaq_end-of-day-data.
2023-02-16 14:34:16 +0000 - dagster - INFO - __ASSET_JOB - 91cb0a21-d1d0-4a94-896f-da41410b8e52 - airbyte_sync_b9e15 - op 'airbyte_sync_b9e15' did not fire outputs {'nasdaq_end_of_day_data'}
2023-02-16 14:34:16 +0000 - dagster - ERROR - __ASSET_JOB - 91cb0a21-d1d0-4a94-896f-da41410b8e52 - 12712 - airbyte_sync_b9e15 - STEP_FAILURE - Execution of step "airbyte_sync_b9e15" failed.

dagster._core.errors.DagsterStepOutputNotFoundError: Core compute for op "airbyte_sync_b9e15" did not return an output for non-optional output "nasdaq_end_of_day_data"

Stack Trace:
  File "C:\Work\MTM Data Research Unified\data-engineering\data-pipelines\env\lib\site-packages\dagster\_core\execution\plan\execute_plan.py", line 265, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "C:\Work\MTM Data Research Unified\data-engineering\data-pipelines\env\lib\site-packages\dagster\_core\execution\plan\execute_step.py", line 382, in core_dagster_event_sequence_for_step
    for user_event in check.generator(
  File "C:\Work\MTM Data Research Unified\data-engineering\data-pipelines\env\lib\site-packages\dagster\_core\execution\plan\execute_step.py", line 175, in _step_output_error_checked_user_event_sequence
    raise DagsterStepOutputNotFoundError(

I assume I have a configuration and/or dependency wire up error but I can't find it. Any pointers gratefully received. I should also say - the Airbyte connection sync is triggered fine and the Dagster console logs contain a bunch of messages suggesting successful progress.

Hello 👋 It looks like the SQLite connection used by Dagster to log events is not thread-safe. The connection is created in one thread and is being used in another thread, causing the following error message :

sqlalchemy.exc.ProgrammingError: (sqlite3.ProgrammingError) SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 140024476800832 and this is thread id 140024381171456.
(Background on this error at: <https://sqlalche.me/e/14/f405>)

Here's how to reproduce the error:

import concurrent.futures

from dagster import get_dagster_logger, graph, op


def my_function(x):
    logger = get_dagster_logger()
    <http://logger.info|logger.info>(f"Got {x=}")
    return x + 1


@op
def my_op():
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        results = list(executor.map(my_function, range(3)))
    print(results)
    return results


@graph
def my_graph() -> None:
    my_op()


def main() -> None:
    my_graph.execute_in_process()


if __name__ == "__main__":
    main()

I wanted to bump this InputManager not acting as expected error:

dagster._core.errors.DagsterInvalidDefinitionError: input manager with key 'model_input_manager' required by input 'encoders' of op 'transformer_op' was not provided. Please provide a <class 'dagster._core.storage.input_manager.IInputManagerDefinition'> to key 'model_input_manager', or change the required key to one of the following keys which points to an <class 'dagster._core.storage.input_manager.IInputManagerDefinition'>: ['io_manager', 'output_notebook_io_manager']

which comes from the following definitions:

class FixedPathInputManager(InputManager):
    extension: str = ".joblib"

    def _get_path(self, context) -> str:
        <http://context.log.info|context.log.info>(context.resource_config)
        <http://context.log.info|context.log.info>(type(context))
        return UPath(f"{context.resource_config['base_path']}/{context.name}{FixedPathIOManager.extension}")

    def load_input(self, context):
        <http://context.log.info|context.log.info>("in load input")
        if context.upstream_output is None: # input manager
            path = self._get_path(context)
        else:
            path = self._get_path(context.upstream_output)
        with path.open("rb") as file:
            return joblib.load(file)

@input_manager(config_schema={"base_path": Field(str, is_required=False)})
def local_model_fixedpath_input_manager(
    init_context: InitResourceContext,
) -> FixedPathInputManager:
    assert init_context.instance is not None  # to please mypy
    base_path = UPath(
        init_context.resource_config.get(
            "base_path", init_context.instance.storage_directory()
        )
    )
    return FixedPathInputManager(base_path=base_path)

and which is used as so:

local_train_transformer_job = transformer_graph.to_job(
    name="train_transformer_job",
    resource_defs={
        "output_notebook_io_manager": local_output_notebook_io_manager,
        "data_file": current_training_data, 
        "data_type": train_type,
        "encoder_file": encoder_file,
        "model_input_manager": local_model_fixedpath_input_manager,
        "lake_io_manager": local_pandas_parquet_io_manager,
    }
)

According to the docs this definition of an InputManager should be enough? Am i being really stupid and missing something super-obvious?

Hello guys , anyone ever got this :

UserWarning: Error loading repository location dagster_poc:dagster._core.errors.DagsterInvalidDefinitionError: resource with key 'dbt' required by op 'run_dbt_05b60' was not provided. Please provide a <class 'dagster._core.definitions.resource_definition.ResourceDefinition'> to key 'dbt', or change the required key to one of the following keys which points to an <class 'dagster._core.definitions.resource_definition.ResourceDefinition'>: ['io_manager']

? I have no clue on how to fix that .

How do Dagster and tools like Segment (Twilio) work together, or is there too much overlap?

Hi, Is possible update the selection of a previous "define_asset_job" created? My use case is: I have a dbt project with multiple models and I have "N selection rules" for these models. So I am thinking that I will need to create N "jobs" with "define_asset_job" by N "selection rules". So I was thinking only create a job and update the selections rules on the fly. is possible?

Is there any documentation out there on how to configure a job that materializes a group of assets who are dependent on the output of an asset not materialized by the job?

Is there a way to get the previous partition_key from context in an asset?

Hello! I have a (hopefully) quick question. I want to materialize an asset on a recurring basis so I am using define_asset_job in conjunction with ScheduleDefinition. My asset requires some resources like a db cursor and requests session. How can I pass these resources to it using define_asset_job?

Hey there I was wondering if its possible to dynamically schedule an op in the future. For example say I want to run this op at 3pm tommorow as a one off.

Hi, I'd like to pass dbt variables to dbt_cli_resource based on Dagster run_id. Run_id is a runtime variable and context is not available at declaration time. What is the best way to accomplish it? Code mockup for my definitions: vars = get_dbt_vars(context.run_id) DBT_CONFIG = {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR, "target": DBT_TARGET, vars: vars} resource_def = { "dbt": dbt_cli_resource(DBT_CONFIG) }

The problem above with input managers seems to be a problem with ops sharing resources. I define a iomanager and use it in an op with a

io_manager_key

. Then I use it in another

Out

in another op, and this is where things blow up. Here is my code:

target_extractor_op = define_dagstermill_op(
    name="target_extractor_op",
    notebook_path=file_relative_path(__file__, "../notebooks/target_extractor.ipynb"),
    output_notebook_name="output_target_extractor",
    outs={"target": Out(pd.DataFrame, io_manager_key="lake_io_manager")},
    ins={"df_train": In(pd.DataFrame)}
)


local_target_extractor_job = target_extractor_graph.to_job(
    name="target_extractor_job",
    resource_defs={
        "output_notebook_io_manager": local_output_notebook_io_manager,
        "training_data": current_training_data,
        "lake_io_manager": local_pandas_parquet_io_manager,
    }
)

transformer_op = define_dagstermill_op(
    name="transformer_op",
    notebook_path=file_relative_path(__file__, "../notebooks/transform.ipynb"),
    output_notebook_name="output_transform",
    outs={"transformed_data": Out(pd.DataFrame, io_manager_key="lake_io_manager")},
    ins={"df": In(pd.DataFrame), "encoders": In(dict), "datatype": In(str)}
)


local_train_transformer_job = transformer_graph.to_job(
    name="train_transformer_job",
    resource_defs={
        "output_notebook_io_manager": local_output_notebook_io_manager,
        "data_file": current_training_data, 
        "data_type": train_type,
        "encoder_file": encoder_file,
        "lake_io_manager": local_pandas_parquet_io_manager,
    }
)

This code blows up on load with this error:

dagster._core.errors.DagsterInvalidDefinitionError: io manager with key 'lake_io_manager' required by output 'transformed_data' of op 'transformer_op'' was not provided. Please provide a <class 'dagster._core.storage.io_manager.IOManagerDefinition'> to key 'lake_io_manager', or change the required key to one of the following keys which points to an <class 'dagster._core.storage.io_manager.IOManagerDefinition'>: ['io_manager', 'output_notebook_io_manager']

Why would it work for the first (target-extractor) op/graph and not the second? All i want to do is to be able to write custom parquet files for some of my ops...is there some type based uniqueness to ops? something else?

Hello Dagster! We have a weird symptom on our scheduled jobs: • we noticed that some of our scheduled jobs have ‘starting’ state but never actually run for > 6 days. We want to know if this kind of issue is only specific to us or happen to others as well.

I currently have a dynamically generated list of values that i'd like to map down through a pipeline of ops until it reaches a

.collect

call eventually, but i'd like to have a limit of only one of those pipelines within a job running at a time. For a visual, in the example photo below i'd like the first mapping index of downstream ops to run and finish execution before the next mapping index is started. I tried per-op prioritization but it seems (correct me if i'm wrong) that it's a prioritization of launching the op, not actually the complete execution of it. Is there an easy way to offer concurrency limits on a per-mapping index baseline, or some ideas about possible method of going about this problem?

Hi, I am trying to integrate in-house applications as part of dagster workflow. A way I am currently considering is that an ops submits a job to celery workers through message queue. When a celery worker completes the task, it submit a result metadata to another queue where a dagster sensor can run on and returns another run request. Is this a possible workflow using the dagster or am I going down to the wrong path?

What's the best approach to handling holidays in a partitioned asset? I created an asset like the following below, should I raise an exception on a holiday? or have it pass along a bool?

@asset(partitions_def=my_partitions_def)
def fed_holiday(context) -> bool:
    return is_fed_holiday(context.partition_key)

I have a custom parquet IO manager that extends

UPathIOManager

than can provide

pandas.DataFrame

or

pyarrow.Table

This generally works fine and I can switch back and forth, but when I have a downstream asset that requires multiple upstream partitions, I get type mismatch error i.e.:

DagsterTypeCheckDidNotPass

. The only way I can get it to work is to have the downstream asset use

upstream: Dict[str, X]

where

X

is the type annotation on the IO manager’s

def load_from_path(...) -> X

method. I can’t seem to find the way to annotate my methods to make this dynamic.

Can asset sensors initiate a run for each materialization of the observed asset since the last sensor tick? After experimenting with them a bit, it seems like the asset sensor will only initiate a run for the latest materialization of the observed asset. I turned off my asset sensor, materialized the observed asset 3 times, then turned my asset sensor on again and the next tick only launched 1 job run when I was expecting 3 job runs. Is this expected?

Hi, is it possible to do conditional branching on subgraphs, similar to this example with ops? https://docs.dagster.io/concepts/ops-jobs-graphs/graphs#with-conditional-branching My use case is that I need to export data to one of several possible third-party services depending on the account the job is being executed for. So far I've assumed there would be one subgraph implementation for each service. Is there a better approach I should try instead? Thank you!