Hi, I’m trying to write a custom

io_manager

that saves the output of a partitioned asset and behaves nicely if a non-partitioned downstream asset asks for the output (it should return the latest available partition), but I got a bit lost on what to look for in the

InputContext

object. I tried to decipher the relevant methods on

UPathIOManager

but couldn’t really grasp what’s going on with the partition keys (comments are confusing for me, I even sense some typos there, but cannot be sure…). How should I decide whether the downstream asset is partitioned so

load_input

should return the specific partition or non-partitioned therefore

load_input

should return the latest available?

hoping some on can explain if its possible to configure configured ops: i have an op that i want to be able to reuse -- which takes a query, runs it in big query, then returns a dataframe

@op(required_resource_keys={"rsrc_bigquery_runner"},
    config_schema={'qry_str': str,
                   'filter_func': Field(Any, default_value=None)}  # this function is meant to allow further play on the dataframe
)
def op_bigq_adjust_df_results(context: OpExecutionContext) -> pl.DataFrame:
	qry_str = context.op_config['qry_str']
	filter_func = context.op_config['filter_func']
	df: pl.DataFrame = context.resources.rsrc_bigquery_runner(qry_str).bigquery_to_df()
	if filter_func is None:
		return df
	else:
		filtered_df = filter_func(df)
		get_dagster_logger().info(
			f'ADJUSTED DATAFRAME SHAPE: {str(filtered_df.shape)} columns: {str(filtered_df.shape[1])}, rows: {str(filtered_df.shape[0])}')
		return filtered_df

then i want to be able to use this op in it's configured form either at the top of a job (no dependency) or somewhere else (with dependency of other ops) so here would be initial configuration of the op - which leaves it open for further configuration:

@configured(configurable=op_bigq_adjust_df_results, config_schema={"query": str, "placeholder": str})
def checking_configured(context, SOME_UPSTREAM_OP):  # SOME_UPSTREAM_OP would provide me with a replacement string for dynamic sql
   replacement = SOME_UPSTREAM_OP
   query = context.op_config['query'].replace(
                                              placeholder=context.op_config['placeholder'],
                                              replacement=replacement)
   return {"qry_str": query, "qry_type": "DQL"}  # this would then be passed to the above base op (op_bigq_adjust_df_results)

and then i would have a third op where i pass the actual values to the above configured op:

x = configured(configurable=checking_configured, name='x')(
	{"query": QRY_BIGQ,
	 "placeholder": '$placeholder$',
	 }
)

i already have a bigquery resource, which has all sorts of methods - but i'd like to see stuff on the dagit chart. and id like to turn some configured ops into assets THANK YOU!

ill be honest some of these concepts are over my head and it would be great if there were tutorials on iomanagers, project structure etc. 😕

I'm having some difficulties using the

execute_job

to test my jobs. I'm using databricks step launcher, and it requires a reconstructable pipeline, so I'm unable to use

job.execute_in_process

. From the docs I'm doing:

instance = DagsterInstance.get()
execute_job(reconstructable(my_job), run_config=config, instance=instance)

But I get this error:

dagster._check.CheckError: Failure condition: Unexpected return value from child process <class 'collections.ChildProcessStartEvent'>

Stack Trace:
  File "/opt/conda/envs/dagster_env/lib/python3.9/site-packages/dagster/_core/execution/api.py", line 991, in pipeline_execution_iterator
    for event in pipeline_context.executor.execute(pipeline_context, execution_plan):
  File "/opt/conda/envs/dagster_env/lib/python3.9/site-packages/dagster/_core/executor/multiprocess.py", line 240, in execute
    event_or_none = next(step_iter)
  File "/opt/conda/envs/dagster_env/lib/python3.9/site-packages/dagster/_core/executor/multiprocess.py", line 364, in execute_step_out_of_process
    check.failed("Unexpected return value from child process {}".format(type(ret)))
  File "/opt/conda/envs/dagster_env/lib/python3.9/site-packages/dagster/_check/__init__.py", line 1687, in failed
    raise CheckError(f"Failure condition: {desc}")

Am I missing something?

A question about combining an asset with

output_required=False

with a

FreshnessPolicy

and `asset_reconciliation_sensor`: if the asset job is written to only materialize (

yield

an

Output

) when some source data on the internet changes, but the

FreshnessPolicy

is set to "1 hour", what happens when the last materialization was 10 hours ago but the source data hasn't changed, yet?

Hi, I am trying to run Pyspark locally from within a job. It works locally, but once I deploy dagster in k8s using helm, I get an error

Java gateway process exited before sending its port number

. It works locally, but not on kubernetes and I am out of ideas. I wanted to try https://github.com/dagster-io/dagster/issues/2748, but all links are broken

Dagster question: I have a machine learning pipeline, and I want to run multiple versions of the model/features In my mind I expect to have to write a custom asset for each version of the features, but thereafter the code is identical for all versions - is there a good way of doing this in Dagster?

Hi first post. I am having a problem with a dagster K8s user deployment. On a localhost dagit I can run the code, and the repo loads fine on the production instance (with the expected config from a yaml file). However upon trying to execute the code on the prod instance of dagster I recieve the following error upon inspecting the errored pod logs:

Traceback (most recent call last):
  File "/usr/local/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.10/site-packages/dagster/_cli/__init__.py", line 48, in main
    cli(auto_envvar_prefix=ENV_PREFIX)  # pylint:disable=E1123
  File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1130, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1055, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1657, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1657, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.10/site-packages/click/core.py", line 760, in invoke
    return __callback(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/dagster/_cli/api.py", line 61, in execute_run_command
    DagsterInstance.from_ref(args.instance_ref)
  File "/usr/local/lib/python3.10/site-packages/dagster/_core/instance/__init__.py", line 472, in from_ref
    run_coordinator=instance_ref.run_coordinator,
  File "/usr/local/lib/python3.10/site-packages/dagster/_core/instance/ref.py", line 437, in run_coordinator
    return self.run_coordinator_data.rehydrate() if self.run_coordinator_data else None
  File "/usr/local/lib/python3.10/site-packages/dagster/_serdes/config_class.py", line 93, in rehydrate
    raise DagsterInvalidConfigError(
dagster._core.errors.DagsterInvalidConfigError: Errors whilst loading configuration for {'max_concurrent_runs': Field(<dagster._config.source.IntSourceType object at 0x7effa69b4dc0>, default=@, is_required=False), 'tag_concurrency_limits': Field(<dagster._config.config_type.Noneable object at 0x7effa02f4370>, default=@, is_required=False), 'dequeue_interval_seconds': Field(<dagster._config.source.IntSourceType object at 0x7effa69b4dc0>, default=@, is_required=False)}.
    Error 1: Received unexpected config entries "['dequeue_num_workers', 'dequeue_use_threads']" at the root. Expected: "['dequeue_interval_seconds', 'max_concurrent_runs', 'tag_concurrency_limits']."

I am confused by this as I do not specify anywhere in my code the "unexpected" config entries .So, I was wondering if anyone has seen and dealt with a similar error before and what they did to fix it?

Hi, I’m new to orchestration tools like Dagster and I’m trying to replicate one of our Fivetran Custom Connector data pipelines as a proof of concept. I have managed to get the data I want from the API and have it in a python dict. I then use json.dumps on the dict and output it in my asset. I have assigned the gcs_pickle_io_manager to the asset and it is loading the file successfully into my GCS bucket. The issue is then when I try to load it into Snowflake. It has a weird character at the beginning and end of the JSON string which is causing it to fail. Depending on which text editor I open it with it changes, but I think it is b-, which indicates that it is a byte string? How do I create the JSON file in the GCS bucket without the characters in the front? There seems to be limited documentation on the IO manager, so not sure if there are options there to change? Thanks!

Hello, I have two questions related to the dagster-k8s retry code https://github.com/dagster-io/dagster/blob/754e4aabb2fe5ac1969d46591ff7e5ff0d9a7b43/python_modules/libraries/dagster-k8s/dagster_k8s/client.py#L81: - It seems it was chosen to not retry all k8s api calls. Is there a specific reason for this? This seems strange to me since all of these calls can potentially encounter network issues. - Did the devs consider using a retry library such as tenacity or expand the current retry functionality? You can currently only retry at a specific interval, and there is no way to, for example, increase the wait interval for each retry attempt.

It looks like the dagster-pandera library is not updated on pypi. Is there a change in way of working? https://pypi.org/project/dagster-pandera/1.0.5/ Posted in #random

Hi, I would like to create a Kubernetes Pod which contains multiple containers. Can I use the execute_k8s_job function to do so? The

image

parameter is not optional, so this leads me to believe that the Pod will only have one container based on that image but then I also see that I can specify

pod_spec_config

which tells me that I can define a Pod with multiple containers, like:

apiVersion: batch/v1
kind: Job
metadata:
  name: hello
spec:
  template:
    # This is the pod template
    spec:
      containers:
      - name: hello
        image: busybox:1.28
        command: ['sh', '-c', 'echo "Hello, Kubernetes!" && sleep 3600']
      restartPolicy: OnFailure
    - name: world
    .....
    # The pod template ends here

but then I'm not sure where the

image

parameter fits in, since each container has its image specified in the Pod Specification. So which is the case? Are there any examples on this?

I've been experimenting with Dagster a bit and have a high-level question related to partitioning. It seems that partition_configs and partition definitions with jobs and assets, respectively, can in practice accomplish the same thing. What are the advantages and disadvantages of using one over the other? I will have partition dependencies, backfills up to 60k in size, and will be running on kubernetes. Thanks for any thoughts.

Hello, sorry for the repost. But we are trying to migrate to the newest dagster version and this is the only problem we have in the migration. Can someone take a look? Thank you 🙏 https://dagster.slack.com/archives/C01U954MEER/p1674423546799289

Hello, Sorry for the beginner question but I couldn't find this in the docs. I have an op that returns two outputs. However the log shows that the second output just overwrites the first output on disk, so in this example both "model" and "metrics" point to to the same "metrics" file on disk as "model" is overwritten. Any ideas?

@op(out={"model": Out(), "metrics": Out()})
def train(dm):

    res = sc.run_training(dm)
    model = res.get_model("logreg")
    metrics = res.get_model_metrics("logreg")

    return (
        Output(model, output_name="model"),
        Output(str(metrics), output_name="metrics"),
    )

Bumping this in case anyone knows why I'm running into this issue with Run Monitoring https://dagster.slack.com/archives/C01U954MEER/p1674259656543229

Hey – Has anyone experimented with dynamic tag concurrency limits? I maintain a pool of connections to a 3rd party resource and have no guarantees on how many of those connections will be available during runtime.

Hi Dagster! I have a project with about 50 repositories. I observed that before each execution the whole code is loaded, which can take a considerable time (1 minute in my case). I think that this loading allows to detect possible errors before the execution. However, in my case it slows down the process a lot. Is there a solution to prevent this ? Would it be possible to make this code loading optional?

Hello, getting GraphQL errors in Sensors and Backfills tabs. mySQL 8.0.31 hosted on GCP. Runs fine on local docker container however looks like there is an issue with 8.0.31-google.

Operation name: SingleSensorQuery

Message: Invalid version: '8.0.31-google'

Path: ["sensorOrError","sensorState","runs"]

Locations: [{"line":28,"column":9}

Is there a way to set up a custom IOManager to act as a context manager? I’m writing to a SQL database and would like to have some safety around opening and closing database connections. With

@resource

, I can use

@contextmanager

and things just work; is that true of

@io_manager

as well?

How is the

@asset

decorator’s

io_manager_key

parameter supposed to work? I can’t seem to get it to recognize a custom IOManager even with a very minimal case (code in thread)

Is there a way of having a master project that reads in "child" projects? For example, we are setting up master repo. We have N developers making their own repos (fully self-contained). Is there a way to read those as subprojects without having to rewrite the master repo?

What's the best way to create a table (our table is in BigQuery) as an asset? We want to create assets for our tables and materialise them when new data is added to them. Anyone here tried this?

Where can I find good tutorials on using Dagster besides dagster.io?

The API is evolving rapidly and I'm struggling to find updated ones that are lengthly and descriptive

Does dagster support loading code from a git repo or s3 storage instead of putting it into a docker image?

does anyone know how you can pass a dependency (another op) into a configuration of a reusable op? for example

x = configured(reusable_op, name='x')({"whatever_is_configured_in_the_reusable_op":'some_arg')

how can i set an op above this one without changing the reusable op?

hoping it has something to do with this

Hi all -- In unit tests, is there an example for mocking partitions that use resources? More details in 🧵

Hello, is there ability to disable or override default stdout log handler. For example if I provide my custom log handler that outputs to stdout. It prints duplicate log lines with different formats. `dagster.yaml`:

...
python_logs:
  python_log_level: INFO
  dagster_handler_config:
    handlers:
      dagster:
        class: logging.StreamHandler
        level: INFO
        stream: <ext://sys.stdout>
        formatter: myFormatter
    formatters:
      myFormatter:
        format: "%(message)s"
        class: pythonjsonlogger.jsonlogger.JsonFormatter
...

Console logs:

...
{"message": "single_pod_job - bc6aee2e-79bc-4225-9d09-8790a2da5647 - multiply_the_word - Multiplying", "dagster_meta": {...}
2023-01-24 15:22:24 +0000 - dagster - INFO - single_pod_job - bc6aee2e-79bc-4225-9d09-8790a2da5647 - multiply_the_word - Multiplying
...