Anyone written a Git hook to check all pipelines / jobs that have been edited have run successfully?

Hello, I'm new to Dagster. I'd like to ask whether there is a way to implement a solution similar to Airflow's HttpSensor? Thank you.

Hello. I'm new to Dagster and started getting a exit code -9 when starting a pipeline on my repo using the UI. Now when I start my docker container it doesn't even load the repo at all. I get the following error

dagster.serdes.ipc.DagsterIPCProtocolError: Process exited with return code -9 while waiting for events

. Would anyone be able to explain what this means?

hi guys, regarding grpc server: there could be a set of hosts (with same repo) sits behind the host domain, is that ok? (I think another way to ask this question is: whether the repo is stateless)

- grpc_server:
      host: the.domain.to.the.repo
      port: 4266
      location_name: "my_grpc_server"

Hi Dagster team! I am getting this error several times. I am trying to launch a pipeline from dagit I got this error:

Traceback (most recent call last):

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\poll_compute_logs.py", line 61, in <module>

execute_polling(sys.argv[1:])

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\poll_compute_logs.py", line 53, in execute_polling

with open(ipc_output_file, "w"):

FileNotFoundError: [Errno 2] No such file or directory: 'C:\\Users\\s4957336\\AppData\\Local\\Temp\\1\\tmpo_zr47w6\\execute-windows-tail-15bc135d998546caa23116bbff00ceb1'

2021-06-14 02:33:51 - dagster - ERROR - Mexico_CTMS_Corp - a1cb3fdf-19ef-4676-a6b0-b3c7610ec0f9 - 22036 - PIPELINE_FAILURE - Execution of pipeline "Mexico_CTMS_Corp" failed. An exception was thrown during execution.

Exception: Timed out waiting for tail process to start

Stack Trace:

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\api.py", line 762, in pipeline_execution_iterator

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

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\executor\in_process.py", line 38, in execute

yield from iter(

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\api.py", line 841, in __iter__

yield from self.iterator(

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\plan\execute_plan.py", line 72, in inner_plan_execution_iterator

active_execution.verify_complete(pipeline_context, step.key)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\storage\compute_log_manager.py", line 56, in watch

yield

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\storage\local_compute_log_manager.py", line 51, in _watch_logs

yield

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\compute_logs.py", line 31, in mirror_stream_to_file

yield pids

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\compute_logs.py", line 75, in tail_to_stream

yield pids

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\compute_logs.py", line 104, in execute_windows_tail

raise Exception("Timed out waiting for tail process to start")

I am not sure why dagster is looking into this path 

FileNotFoundError: [Errno 2] No such file or directory: 'C:\\Users\\s4957336\\AppData\\Local\\Temp\\1\\tmpo_zr47w6\\execute-windows-tail-15bc135d998546caa23116bbff00ceb1'

 what could I do? I am using windows 10, 64 bit. Conda enviroment. Dagster 0.12.2 version. I got a timeout several times randomnly. I configured the backend in a postgres Instance in dagster.yml file.

Trying to test a schedule with the following code, but I get the error below.

@pipeline(
    mode_defs=[
        mode_def_prod,
        mode_def_prod_coll,
    ],
)
def collective_pipeline():
    ...


@schedule(
    pipeline_name="collective_pipeline",
    cron_schedule="0 7 * * *",
    execution_timezone="UTC",
    mode='prod_coll'
)
def coll_daily_schedule():
    return {"resources": {"io_manager": {"config": {"s3_bucket": "dagster-wdt", "s3_prefix": "stage/io"}}}}


def test_coll_daily_schedule():
    run_config = coll_daily_schedule()
    assert validate_run_config(collective_pipeline, run_config)

error when I use

pytest

in commandline:

dagster.core.errors.DagsterInvalidConfigError: Error in config for pipeline collective_pipeline
Error 1: Received unexpected config entries "['s3_bucket', 's3_prefix']" at path root:resources:io_manager:config. Expected: "['base_dir']."

One strange solution I have found is to switch the order of the mode definitions in the pipeline so that

mode_def_prod_coll

comes first. Then the test passes. But in my mind, this shouldn’t happen because I’m specifically calling out the name of the mode I want to use in the decorator of the schedule

prod_coll

Any suggestions for this please 🙏? https://dagster.slack.com/archives/C01U954MEER/p1626819763322900

Hi team. I am trying to run dagster-daemon but I got a error realted with GRPC server. I kill all the process related to dagster on my machine Windows 10 however I am still getting the same error. Could you provide me any advice? thank you in advance. I am very curious about the last line. I didn't start other daemon in any place. I can only run pipeline using API but I can't use dagit and/or daemon. The appears after I update dagster version to 0.12.2 from 0.11.1.

I'm running my solids locally via multiprocess execution, and now I want to use multiprocess to improve the performance of Pymongo within a final solid which will always be running on its own. I'm not super familiar with multiprocess, but I'd assume that that solid is running in a single process and therefore is unable to use multiprocessing itself ? But this final solid will never be executed at the same time as another, so might it be possible to revert to in_process execution for the final solid? Feel like the answer to this will help me understand python / multi-processing better in general as much as Dagster. Would really appreciate if someone could help me understand this better.

Hey, just tried setting up a new workspace and now when I'm running dagit I get:

No module named dagster.grpc

I installed

dagster==0.12.2 dagit==0.12.2 dagster-graphql==0.12.2

Running

python repo.py

works fine but for some reason dagit can't load it. Any ideas why?

Hi all. I’m experimenting with using an asset sensor to trigger a cross-repository pipeline and I’m wondering how I can send alerts (ie, slack) if the sensor itself fails exceptionally. Any ideas how/where to look further?

Hi, Dagster Team. I found [docs] link in source code doc is black color and it hard to see. The link that I open is https://docs.dagster.io/_modules/dagster/core/execution/context/hook#HookContext. Found on both Chrome and Firefox. Not sure everyone found this issue like me?

Hello everyone, I am having some issues using the Avanced Deployment and wondering if anyone could help. At the moment the issue is that I don't know how to pass the redis password to the celery workers. I'm trying the following:

runLauncher:
  type: CeleryK8sRunLauncher
  config:
    celeryK8sRunLauncher:
      configSource:
        redis_password:
          valueFrom:
            secretKeyRef:
              name: wot-dagster-redis
              key: redis-password

redis:
  enabled: true
  internal: true
  host: "wot-dagster-redis-master"

by using the redis>interal: true the wot-dagster-redis gets generated automatically as far as I understand and the "redis-password" is set inside. I'm then trying to give that password to the celeryK8sRunLauncher as otherwise the celery pods return:

[2021-07-22 12:40:27,366: ERROR/MainProcess] consumer: Cannot connect to <redis://wot-dagster-redis-master:6379/0>: Authentication required..

Any help welcome 🙂 Gabe

I've setup several schedules and sensors. Each time I set one up I also need to open Dagit and turn the schedule/sensor on. Is there a way to have the schedules/sensors on by default?

Getting an “Unexpected GraphQL error” in the dagit UI while trying to view the ‘Runs’ Page. Erro info:

Operation name: RunsRootQuery

Message: list index out of range

Path: ["pipelineRunsOrError","results",2,"canTerminate"]

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

Stack Trace:
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/executor.py", line 452, in resolve_or_error
    return executor.execute(resolve_fn, source, info, **args)
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/executors/sync.py", line 16, in execute
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/schema/pipelines/pipeline.py", line 274, in resolve_canTerminate
    return graphene_info.context.instance.run_coordinator.can_cancel_run(self.run_id)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/run_coordinator/queued_run_coordinator.py", line 105, in can_cancel_run
    return self._instance.run_launcher.can_terminate(run_id)
  File "/usr/local/lib/python3.7/site-packages/dagster_aws/ecs/launcher.py", line 138, in can_terminate
    status = self.ecs.describe_tasks(tasks=[arn], cluster=cluster)["tasks"][0]["lastStatus"]

I am running dagster on the experimental EcsRunLauncher in AWS ECS Fargate. Version 0.12.1. Prior to this error, I terminated a couple of long-running runs that were in the middle of execution but had errored on previous steps.

Trying to test a solid that accesses

mapping_key

from context. Getting error:

def get_mapping_key(self) -> Optional[str]:
        """
        Which mapping_key this execution is for if downstream of a DynamicOutput, otherwise None.
        """
>       return self._step_execution_context.step.get_mapping_key()
E       AttributeError: _step_execution_context

Test code is pretty simple:

context = build_solid_context(
        config={...})

    data = ...

    export_data(context, data)

Hi team, I have a question on Dagit helm chart release (0.11.15) pulling postgres images. While deploying dagit component, the container is still trying to pull postgres 9.6 image from docker public repo despite we have set postgres “enabled: false” and image:registry pointing to our private ECR. Is this a known issue or if anyone knows a workaround please?

Hey all, is there an easy way to use variables at different steps of Dynamic outputs? I have something like so:

def pipeline():
  test = generator() # generates dynamic output
  other_solid_output = solid()
  output1, output2 = test.map(process1).map(process2)
  more_output = new_solid(output1, other_solid_output)
  last_process(output2, more_output)

I get the error that Dynamic Output must be mapped or collected. I tried using map with these too but that just returns syntax error. Is there any feasible way to make this work? Without changing the logic of the current solids

Hi team, For most of our pipelines, we are dynamically generating the PartitionSets based on metadata from DB. Each PartitionSet targets a particular aggregate on a data source and hence the run config changes for each of them. More discussion on this design here Currently, I am naming the partition based on the aggregation and group by keys. All the different partition sets are available on the backfill UI, but the names are quite long and a bit confusing. It also will be quite hard to search them once we have too many partition sets. Is there any plan to support tags for the PartitionSet definitions? AFAIK, this is not available right now.

Hi, How do we get solid processing time via hook?

Hi folks, We are currently migrating from the 0.11.4 to 0.12.2 and we are experiencing the following issue while trying to execute

dagster instance migrate

. It seems that it can't reindex assets for some reasons. Do you have any idea to fix this ?

root@02e3ef89f539:/home/src/dagster# dagster instance migrate
$DAGSTER_HOME: /home/src/dagster

Updating run storage...
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 7cba9eeaaf1d -> ddcc6d7244c6, add column mode
INFO  [alembic.runtime.migration] Running upgrade ddcc6d7244c6 -> 7b8304b4429d, extract asset_keys index columns
Starting data migration: add_mode_column
Querying run storage.
  0%|                                                                                                  | 0/36 [00:00<?, ?it/s]100%|████████████████████████████████████████████████████████████████████████████████████████| 36/36 [00:00<00:00, 445.62it/s]

Finished data migration: add_mode_column
Updating event storage...
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
Starting reindex: asset_key_index_columns
Querying asset keys.
Found 94 assets to reindex.
  0%|                                                                                                  | 0/94 [00:00<?, ?it/s]
  0%|                                                                                                  | 0/94 [00:00<?, ?it/s]

Traceback (most recent call last):
  File "/usr/local/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.7/site-packages/dagster/cli/__init__.py", line 48, in main
    cli(auto_envvar_prefix=ENV_PREFIX)  # pylint:disable=E1123
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1137, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1062, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1668, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1668, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 763, in invoke
    return __callback(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/dagster/cli/instance.py", line 46, in migrate_command
    instance.upgrade(click.echo)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/instance/__init__.py", line 557, in upgrade
    self._event_storage.reindex_assets(print_fn=print_fn)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/event_log/sql_event_log.py", line 434, in reindex_assets
    self._apply_migration(migration_name, migration_fn, print_fn, force)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/event_log/sql_event_log.py", line 421, in _apply_migration

    migration_fn()(self, print_fn)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/event_log/migration.py", line 133, in migrate_asset_keys_index_columns
    tags = event.dagster_event.step_materialization_data.materialization.tags
AttributeError: 'NoneType' object has no attribute 'dagster_event'

Many thanks and have a nice weekend ! Nicolas

Hello again, I am still struggling with the advanced deployment with Celery. I think I have a similar error to this person: https://github.com/dagster-io/dagster/discussions/4282 The following configuration:

imagePullSecrets:
  - name: prod-wot-registry-secrets

dagster-user-deployments:
  enabled: true
  imagePullSecrets:
    - name: prod-wot-registry-secrets
  deployments:
    - name: "wot-dagster-repository"
      image:
        repository: "<path-to-my-repo-on-the-registry>"
        tag: latest
        pullPolicy: Always
      dagsterApiGrpcArgs:
        - "-f"
        - "/app/repository/repository.py"
      port: 3030
      envSecrets:
        - name: prod-wot-storage-secrets

runLauncher:
  type: CeleryK8sRunLauncher

redis:
  enabled: true
  internal: true
  host: "wot-dagster-redis-master"
  usePassword: false

flower:
  enabled: true

dagsterDaemon:
  runCoordinator:
    enabled: true

and get the following error:

dagster.core.errors.DagsterInvalidConfigError: Errors whilst loading configuration for {'instance_config_map': Field(, default=@, is_required=True), 'postgres_password_secret': Field(, default=@, is_required=False), 'dagster_home': Field(, default=/opt/dagster/dagster_home, is_required=False), 'load_incluster_config': Field(, default=True, is_required=False), 'kubeconfig_file': Field(, default=None, is_required=False), 'broker': Field(, default=@, is_required=False), 'backend': Field(, default=rpc://, is_required=False), 'include': Field(, default=@, is_required=False), 'config_source': Field(, default=@, is_required=False), 'retries': Field(, default={'enabled': {}}, is_required=False)}.
    Error 1: Post processing at path root:instance_config_map of original value {'env': 'DAGSTER_K8S_INSTANCE_CONFIG_MAP'} failed:
dagster.config.errors.PostProcessingError: You have attempted to fetch the environment variable "DAGSTER_K8S_INSTANCE_CONFIG_MAP" which is not set. In order for this execution to succeed it must be set in this environment.

Stack Trace:
  File "/usr/local/lib/python3.9/site-packages/dagster/config/post_process.py", line 77, in _post_process
    new_value = context.config_type.post_process(config_value)
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 42, in post_process
    return str(_ensure_env_variable(cfg))
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 16, in _ensure_env_variable
    raise PostProcessingError(

    Error 2: Post processing at path root:postgres_password_secret of original value {'env': 'DAGSTER_K8S_PG_PASSWORD_SECRET'} failed:
dagster.config.errors.PostProcessingError: You have attempted to fetch the environment variable "DAGSTER_K8S_PG_PASSWORD_SECRET" which is not set. In order for this execution to succeed it must be set in this environment.

Stack Trace:
  File "/usr/local/lib/python3.9/site-packages/dagster/config/post_process.py", line 77, in _post_process
    new_value = context.config_type.post_process(config_value)
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 42, in post_process
    return str(_ensure_env_variable(cfg))
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 16, in _ensure_env_variable
    raise PostProcessingError(

The changelog for 0.12.3 seems to be empty?

guys, i set the following

os.environ["PYTHONWARNINGS"] = "ignore"

but i still can see warnings from the log

res = conn.execute(run_group_query)
/export/content/lid/apps/dagster-web/i001/libexec/dagster-web_3b298aacc6b7e5bd29d8bd9bb36dfcb367618b6abfa16d967bb355b20119311b/site-packages/graphql/backend/core.py:32: DeprecationWarning: The 'root' alias has been deprecated. Please use 'root_value' instead.
  return execute(schema, document_ast, *args, **kwargs)
/export/content/lid/apps/dagster-web/i001/libexec/dagster-web_3b298aacc6b7e5bd29d8bd9bb36dfcb367618b6abfa16d967bb355b20119311b/site-packages/graphql/backend/core.py:32: DeprecationWarning: The 'context' alias has been deprecated. Please use 'context_value' instead.
  return execute(schema, document_ast, *args, **kwargs)
/export/content/lid/apps/dagster-web/i001/libexec/dagster-web_3b298aacc6b7e5bd29d8bd9bb36dfcb367618b6abfa16d967bb355b20119311b/site-packages/graphql/backend/core.py:32: DeprecationWarning: The 'root' alias has been deprecated. Please use 'root_value' instead.
  return execute(schema, document_ast, *args, **kwargs)
/export/content/lid/apps/dagster-web/i001/libexec/dagster-web_3b298aacc6b7e5bd29d8bd9bb36dfcb367618b6abfa16d967bb355b20119311b/site-packages/graphql/backend/core.py:32: DeprecationWarning: The 'context' alias has been deprecated. Please use 'context_value' instead.
  return execute(schema, document_ast, *args, **kwargs)
/export/content/lid/apps/dagster-web/i001/libexec/dagster-web_3b298aacc6b7e5bd29d8bd9bb36dfcb367618b6abfa16d967bb355b20119311b/site-packages/dagster/core/storage/runs/sql_run_storage.py:407: SADeprecationWarning: Executing a subquery object is deprecated and will raise ObjectNotExecutableError in an upcoming release.  Please execute the underlying select() statement directly.
  res = conn.execute(run_group_query)

do you know whatelse should i do to get rid of them?

Hey, I'm trying out Dagster and followed the instructions to get it running on a remote host using docker-compose. Dumb question... lets say I want to run a

test.py

script on my local machine and want to hit the dagster instance on my remote host, how do I configure that? I'm sure its something simple that I've missed somewhere in the docs. Thanks!

Hi all, I’m testing my GKE deployment and am wondering how to properly set up with GCS. Which part in the tutorial (https://docs.dagster.io/deployment/guides/kubernetes/deploying-with-helm), should I change so that I use my GCS buckets instead of S3?

guys, do you have plan to support drag & drop solids to create pipelines?

Hi all! Is there a way one can write a scheduled pipeline for wiping all runs/schedules/sensors history older than, for example, one week? Has anyone done that before?

Hi all, I am trying to create a pipeline with a solid that receives a dynamic and regular output. Is this possible with the

map() collect()

system? I am thinking of a flow like the following (I know this isnt how map works, just trying to get the point across):

dynamic_out = get_values() # returns a dynamicOutput
val = get_some_value() # returns an Output
incrimented = incriment(dynamic_out.map(), val)