Hi. I am trying to run dagster (dagit, dagster daemon) on postgres. No docker for the moment. I have run_storage, event_log_storage and schedule_storage all set in the dagster.yaml file. There are 11 tables created in the dagster database, and when runs are performed data shows up in the “runs” table and the event_logs table (for example) in the database. However, the “jobs” table remains empty, even though the dagit console shows 8 jobs (all of which are defined in a workspace.yaml file. How many tables should I see if dagster is “fully” postgres enabled and what should I see in tables (and when) such as the jobs table in the database ? Thank you!

When an Op has one or more inputs, how does it access the context?

Is dagster 0.13.12 updated for GraphQL API?

Hi guys, I would like to know if It is possible to have one qsub for each ops on PBS, instead of one qsub for all the graph ? 🙂

hii Team ,PresetDefinition is not default set configuration in pipeline playground ,it is showing me to click and select Preset can anyone have any idea how directly show these preset value in playground like:

HI I have a custom image that I am trying to get dagster to run using Helm user_deployments with values.yaml that look something like, _args: "_`dagsterApiGrpcArgs: "--python-file" "/path.../repo.py`_"_. The image that I am trying to run has a pybinary built by bazel that srcs dagster_cli.py with dagster dependencies available only to the binary dagster_cli file and repo.py thats generated by the build. This works for the base image that runs the gRPC server because the command passed to the kub pod for the base can utilize that dagster_cli binary with "`Command: dagster Args: api grpc -h ...`". But the executor/runner pod gets a different command using the base python "`/usr/local/bin/python3 -m dagster api execute_run {some_args for repo.py}.`" My question is, Bazel does not pip install the dagster dependencies to the base python, so is there a good way to get the runner to work with only a dagster_cli binary on the image?

Another way of asking the question above is, for build tools like bazel which are deterministic and hermetic in nature, and work to prevent pip installing to the base python, what is the best way to setup a dagster-user-deployment image to get the runner to work?

I am hitting a strange error but I am not sure if it's a bug or a mis-use on my part so wanted to ask here first, as the documentation has not proven to be illuminating on this point. I have a graph that calls several operators which return DataFrames and use an IOManager to write them to parquet:

@graph

def load_itsa_assets():

universe = load_token_universe()

id_map = load_id_map()

joined_df = pd.concat([id_map.set_index('symbol'), universe.set_index('symbol')], axis=1, join='inner')

load_classifications(joined_df['itin'].tolist())

For example:

@op(out={'token_universe': Out(io_manager_key='parquet_io_manager', dagster_type=UniverseDataFrame)})

def load_token_universe(context) -> pd.DataFrame:

universe_cfg = toml.load(UNIVERSE_CFG)

symbols = []

for symbol_cfg in universe_cfg['universe']:

symbols.append({

'symbol': symbol_cfg['symbol'],

'cmc_id': symbol_cfg['cmc_id'],

'effective_date': symbol_cfg['effective_date']

})

df = pd.DataFrame(symbols)

df['effective_date'] = <http://pd.to|pd.to>_datetime(df['effective_date'], utc=True)

return df

but the return types are not DataFrames: they are instead

InvokedSolidOutputHandle

, and I am unable to make regular DataFrame method calls on them. What am I missing? How do I get the actual DataFrame out?

Awkward question: I submitted a bug report five weeks ago for a one-line bug. A week ago I submitted a PR with the one-line fix. Neither has been acknowledged -- it's particularly strange to me that the PR has just been sitting there, and in fact there appears to be a backlog of over 60 open PR's. Is this just because of the holidays, or is this normal?

hii Team i want to use more than one PresetDefinition and ModeDefinition like given Below but i am getting some errors please let me know why i am getting this ??

Hello, I was hoping to find out how to increase/remove the grpc timeout for sensors - we are using a dagster sensor that begins by getting a list of files in a very large directory and I am finding that the grpc deadline is being exceeded before it can yield any runs

Hi guys 🙂 Have a good new year ! I'm struggling with this code :

from dagster import fs_io_manager, job, Any
from dagster_dask import dask_executor
from pathlib import Path
from semaphore_scripts import template_runner
import os
from dagster import Nothing, In, op

@op
def adder_purified(purified_ftp: Any, purified_ftp2: Any, purified_eftp1: Any, purified_eftp2: Any, purified_httpvpublic: Any, purified_httpvpublicnew: Any) -> Any:
    return purified_ftp, purified_ftp2, purified_eftp1, purified_eftp2, purified_httpvpublic, purified_httpvpublicnew

def template_runner_op(
    name="default_name",
    ins=None,
    **kwargs,
):
    """
    Args:
        args (any): One or more arguments used to generate the nwe op
        name (str): The name of the new op.
        ins (Dict[str, In]): Any Ins for the new op. Default: None.

    Returns:
        function: The new op.
    """

    @op(
        name=name,
        ins=ins or {"start": In(Nothing)},
        config_schema={
            "command": str,
            "command_parameter": str,
            "semaphore_audience_version": str,
            "workspace": str
        },
        required_resource_keys={"io_manager"},
        **kwargs
    )
    def _template_runner_op(context):
        workspace: Path = Path(
            os.path.join(context.op_config["workspace"], f'{context.op_config["command"]}-{os.environ["PBS_JOBID"]}'))

        template_runner.TemplateRunner(
            command=context.op_config["command"],
            version=context.op_config["semaphore_audience_version"],
            command_parameter=context.op_config["command_parameter"],
            workspace=workspace,
            skip_signal=True
        ).run(exit=False)

    return _template_runner_op

purify_ftp = template_runner_op(name="purify_ftp")
purify_ftp2 = template_runner_op(name="purify_ftp2")
purify_httpvpublic = template_runner_op(name="purify_httpvpublic")
purify_httpvpublicnew = template_runner_op(name="purify_httpvpublicnew")
purify_eftp1 = template_runner_op(name="purify_eftp1")
purify_eftp2 = template_runner_op(name="purify_eftp2")

partition = template_runner_op(name="partition")
event_session = template_runner_op(name="event_session")

#Business/Export
business_oceanography = template_runner_op(name="business_oceanography")
export_oceanography = template_runner_op(name="export_oceanography")

business_sextant = template_runner_op(name="business_sextant")
export_sextant = template_runner_op(name="export_sextant")

business_seanoe = template_runner_op(name="business_seanoe")
export_seanoe = template_runner_op(name="export_seanoe")

business_archimer = template_runner_op(name="business_archimer")
export_archimer = template_runner_op(name="export_archimer")

export_smoswind = template_runner_op(name="export_smoswind")

@job(
    name='template_runner_job',
    resource_defs={"io_manager": fs_io_manager},
    executor_def=dask_executor
)
def template_runner_job():
    #event = event_session(partition(start=[purify_ftp(), purify_ftp2(), purify_eftp1(), purify_eftp2(), purify_httpvpublic(), purify_httpvpublicnew()]))
    event = event_session(partition(purify_ftp(purify_ftp2(purify_eftp1(purify_eftp2(purify_httpvpublic(purify_httpvpublicnew())))))))

    export_oceanography(business_oceanography(event))
    export_sextant(business_sextant(event))
    export_seanoe(business_seanoe(event))
    export_archimer(business_archimer(event))
    export_smoswind(event)

If i do this i obtain a graph with in-series purify op which is not what i want ( i want them in parallel ) BUT i have no error when launching my graph But if i change the in-series purify to have parallelised purify from this :

event = event_session(partition(purify_ftp(purify_ftp2(purify_eftp1(purify_eftp2(purify_httpvpublic(purify_httpvpublicnew())))))))

To this

event = event_session(partition(start=[purify_ftp(), purify_ftp2(), purify_eftp1(), purify_eftp2(), purify_httpvpublic(), purify_httpvpublicnew()]))

My graph looks great ( cf picture joined ) But unfortunately i obtain this error on launch :

dagster.core.errors.DagsterInvariantViolationError: template_runner_job not found at module scope in file /home1/datawork/semexp/workspace/dagit/semaphore-dagster-config/jobs/process_daily.py.

Stack Trace:
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/core/execution/api.py", line 775, in pipeline_execution_iterator
    for event in pipeline_context.executor.execute(pipeline_context, execution_plan):
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster_dask/executor.py", line 280, in execute
    for future, result in iterate_with_context(raise_execution_interrupts, futures):
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/utils/__init__.py", line 383, in iterate_with_context
    next_output = next(iterator)
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/distributed/client.py", line 4446, in __next__
    return self._get_and_raise()
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/distributed/client.py", line 4437, in _get_and_raise
    raise exc.with_traceback(tb)
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster_dask/executor.py", line 134, in query_on_dask_worker
    execution_plan = create_execution_plan(
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/core/execution/api.py", line 735, in create_execution_plan
    pipeline_def = pipeline.get_definition()
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/core/definitions/reconstructable.py", line 112, in get_definition
    defn = self.repository.get_definition().get_pipeline(self.pipeline_name)
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/core/definitions/reconstructable.py", line 48, in get_definition
    return repository_def_from_pointer(self.pointer)
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/core/definitions/reconstructable.py", line 560, in repository_def_from_pointer
    target = def_from_pointer(pointer)
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/core/definitions/reconstructable.py", line 502, in def_from_pointer
    target = pointer.load_target()
  File "/home/datawork-semaphore-exp/conda/conda_envs/semaphore-scripts-1.0.11/lib/python3.9/site-packages/dagster/core/code_pointer.py", line 231, in load_target
    raise DagsterInvariantViolationError(

So if someone has an idea on what's going on i'm listening to you guys ( and girls ) 🙂 PS: I have another questions 1: When launching my graph on PBS via a dask executor, there is one QSUB opened and each steps of the graphs happend in this qsub, is there a possibility to kill the qsub after each step of the graph then recreate a new QSUB ? 2: Maybe linked to the previous question but i can't track the progression of the different steps in the dagit interface, even if under the hood my steps is done, in the interface it stay as running ( cf screen capture 2 )

Hello everyone 👋 There is one thing in dagster that is still not 100% clear to me... I am using ins/outs to define my ops inputs and outputs. When i try to do the same for graphs/jobs though, this does not seem to work as I would expect. This graph for example:

@graph(
    ins={
        "gcs_list": In(List[str]),
    }
)
def airbus_tristereo_orthorectification_pansharpening(gcs_lists):
    generate_subtasks(gcs_lists).map(orthorectify_pansharpen_tristereo_op)

I would expect dagit to ask me for a parameter gcs_list before starting the job but indeed that's not the case. Is it possible to define job/graph-wide inputs and outputs?

Hi i am running dagster by docker-compose and i am using docker launcher and the workspace yaml is:

load_from:
  # Each entry here corresponds to a service in the docker-compose file that exposes user code.
  - grpc_server:
      host: single_sample_pipeline
      port: 4000
      location_name: "<http://gcr.io/immunai-registry-hub/panacea-ai/immunai-product-single_sample_pipeline:eldan-onbuild3|gcr.io/immunai-registry-hub/panacea-ai/immunai-product-single_sample_pipeline:eldan-onbuild3>"

and on the dagit UI on workspace there are not error all loaded, but i cannot run jobs

what am i missing here?

the command for the grpc is:

dagster api grpc -h 0.0.0.0 -p 4000 -m single_sample_pipeline

in the workspace i get

Hi is there a limitation for the repositories names when it contains dots like image from an official registry gcr.io?

osr the limitation is by the name length?

Hi, I get the following GraphQL error when I click the snapshot link (run id) of a finished run in Dagit (version 0.13.12 of all corresponding packages):

Operation name: PipelineExplorerRootQuery

Message: 'HistoricalPipeline' object has no attribute 'get_external_origin_id'

Path: ["pipelineSnapshotOrError","solidHandles",0,"solid","definition","id"]

Locations: [{"line":261,"column":5}]

Stack Trace:
  File ".../lib/python3.8/site-packages/graphql/execution/executor.py", line 452, in resolve_or_error
    return executor.execute(resolve_fn, source, info, **args)
  File ".../lib/python3.8/site-packages/graphql/execution/executors/sync.py", line 16, in execute
    return fn(*args, **kwargs)
  File ".../lib/python3.8/site-packages/dagster_graphql/schema/solids.py", line 537, in resolve_id
    return f"{self._represented_pipeline.get_external_origin_id()}:{self._solid_def_snap.name}"

It's not bothering me but I didn't find an existing issue on that topic and thought I'd post it here, in case this issue is unknown.

Hi i have found an issue when i set in the environment variable: DAGSTER_CURRENT_IMAGE=gcr.io/xxxx/xxx:latest and in similar in the wrokspace.yml at location_name: gcr.io/xxxx/xxx:latest its does not recognise the jobs in the repositories although its recognise the docker image --> only if i retag the image without "/" sign it works, it annoying to change the official image name any familiar with kind issue?

set the channel topic: 🙋 Ask your question after checking out our documentation! https://docs.dagster.io/

Hi everyone! I am trying to test a partitioned job with

execute_in_process

. The partitioned job was build with:

my_graph = my_job.graph

my_partitioned_job = <http://my_graph.to|my_graph.to>_job(resource_defs=my resources, config=my_partitioned_config)

I do know that having a partitioned job built from a graph of a job is unorthodox (I am only doing because this job is sometimes invoked in both contexts), but it should work right? Well, I am stuck at the following error message:

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/definitions/job.py", line 146, in execute_in_process

return core_execute_in_process(

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/execute_in_process.py", line 64, in core_execute_in_process

event_list = list(_execute_run_iterable)

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/api.py", line 861, in iter

yield from self.iterator(

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/api.py", line 774, in pipeline_execution_iterator

for event in pipeline_context.executor.execute(pipeline_context, execution_plan):

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/executor/in_process.py", line 38, in execute

yield from iter(

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/api.py", line 861, in iter

yield from self.iterator(

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/plan/execute_plan.py", line 69, in inner_plan_execution_iterator

for step_event in check.generator(_dagster_event_sequence_for_step(step_context)):

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/plan/execute_plan.py", line 294, in _dagster_event_sequence_for_step

raise unexpected_exception

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/plan/execute_plan.py", line 195, in _dagster_event_sequence_for_step

for step_event in check.generator(step_events):

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster_databricks/databricks_pyspark_step_launcher.py", line 148, in launch_step

step_run_ref = step_context_to_step_run_ref(

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/execution/plan/external_step.py", line 145, in step_context_to_step_run_ref

return StepRunRef(

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/core/definitions/step_launcher.py", line 38, in new

check.inst_param(recon_pipeline, "recon_pipeline", ReconstructablePipeline),

File "/home/cortez/anaconda3/envs/dagster_0_13_12_spark/lib/python3.9/site-packages/dagster/check/__init__.py", line 189, in inst_param

raise _param_type_mismatch_exception(

dagster.check.ParameterCheckError: Param "recon_pipeline" is not a ReconstructablePipeline. Got <dagster.core.definitions.pipeline_base.InMemoryPipeline object at 0x7f09d637af70> which is type <class 'dagster.core.definitions.pipeline_base.InMemoryPipeline'>.

PS: I am using dagster 0.13.12

@Thomas Mignon Given it’s a yaml file, I don’t think yaml properly supports string concatenation, weirdly enough, but I’ve found that defining a root path in a resource and then using a python

Path

object works pretty well

Hi everyone, I'm struggling to get my dagster instance to execute the scheduled jobs - I'm getting "No run requests returned for scheduler, skipping". How is this resolved? Or how do I see what the issue is? For context, I'm on Windows 10 and was using this code almost exactly (switched solid > op and pipeline > job) on 0.11.11. I recently upgraded to the latest version and I'm not able to get the schedule running. Manually kicking off the jobs using dagit works fine. Any help is appreciated!

I have the following YAML:

ops: 
  purify_ftp: 
    config: 
      command: purify_raw_logs
      semaphore_audience_version: '3.4.9'
      command_parameter: '/home/datawork-semaphore-exp/spool/input/ftp/ftp/2022 ftp -f access.log.20220101.gz'
      workspace: /home1/datawork/semexp/workspace/dagit/running/
  purify_ftp2: 
    config: 
      command: purify_raw_logs
      semaphore_audience_version: '3.4.9'
      command_parameter: '/home/datawork-semaphore-exp/spool/input/ftp/ftp2/2022 ftp2 -f access.log.20220101.gz'
      workspace: /home1/datawork/semexp/workspace/dagit/running/

How can I "normalise" this, by removing hardcoded dates, and have it as its own setting, something like:

day: 01
month: 01
year: 2020
ops: 
  purify_ftp: 
    config: 
      command: purify_raw_logs
      semaphore_audience_version: '3.4.9'
      command_parameter: '/home/datawork-semaphore-exp/spool/input/ftp/ftp/*year* ftp -f access.log.*year**month**day*.gz'
      workspace: /home1/datawork/semexp/workspace/dagit/running/
  purify_ftp2: 
    config: 
      command: purify_raw_logs
      semaphore_audience_version: '3.4.9'
      command_parameter: '/home/datawork-semaphore-exp/spool/input/ftp/ftp2/*year* ftp2 -f access.log.*year**month**day*.gz'
      workspace: /home1/datawork/semexp/workspace/dagit/running/

I don't know if a syntax of this type exist (edited)

Hi, I have an optional feature request: the keyboard shortcut Ctrl + A, which is used to select all text, is not working in the launchpad. This would be nice to have.

Hiya folks, I realise this is probably very low on the development list, but is it possible to change the color of the top bar in dagit? (I'd like to make my production and acceptance environments visually distinct from each other)

Hi, When I update my code (i.e. dagster job) do I need to refresh the repository manually or the daemon in the background can do this for me?

My Dagit instance intermittently hangs up and seems to freeze the networking of the AWS EC2 server. Has anyone else notice this?