dagster #dagster-support

Chris Evans

09/14/2021, 5:11 PM

Hi guys, I am trying to control concurrency of runs via tags (0.12.10). I am able to achieve the expected behavior with the

max_concurrent_runs

param, but for the life of me I can’t seem to get things to work with

tag_concurrency_limits

. Do I have this test setup correctly?

Copy code

run_coordinator:
  module: dagster.core.run_coordinator
  class: QueuedRunCoordinator
  config:
    max_concurrent_runs: 2
    tag_concurrency_limits:
      - key: "test"
        limit: 1

Sebitien

09/14/2021, 9:45 PM

Hi guys, I am currently facing an issue (0.12.10) in my project. I regularly trigger a pipeline using a schedule. My issue is that at each execution of the pipeline, one of the solid should be able to fetch some persistent/global state and, after some computation, update its value for later excecutations to use. In other words I want to have within a solid some buffer shared across executions that are triggered by the scheduler. I have been working on some work around using the metadata of a dummy assetmaterialization that I fetch using context.instant.get_event_records but I was wondering if someone came up with a cleaner way to do ? Thanks in advance !

Cameron Gallivan

09/15/2021, 12:22 AM

I’m trying to troubleshoot this error that causes our docker-compose deployment on an EC2 instance to become inoperable. This is the log from the dagster daemon container:

Cameron Gallivan

09/15/2021, 12:22 AM

Copy code

WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
WARNING:root:Retrying failed database connection
2021-09-14 07:20:20 - SensorDaemon - ERROR - Sensor daemon caught an error for sensor dragen_postprocessing_sort_sqs_sensor : dagster_postgres.utils.DagsterPostgresException: too many retries for DB connection

Stack Trace:
  File "/opt/venv/lib/python3.8/site-packages/dagster/daemon/sensor.py", line 230, in execute_sensor_iteration
    yield from _evaluate_sensor(
ERROR:SensorDaemon:Sensor daemon caught an error for sensor dragen_postprocessing_sort_sqs_sensor : dagster_postgres.utils.DagsterPostgresException: too many retries for DB connection

Stack Trace:
  File "/opt/venv/lib/python3.8/site-packages/dagster/daemon/sensor.py", line 230, in execute_sensor_iteration
    yield from _evaluate_sensor(
  File "/opt/venv/lib/python3.8/site-packages/dagster/daemon/sensor.py", line 114, in __exit__
    self._write()
  File "/opt/venv/lib/python3.8/site-packages/dagster/daemon/sensor.py", line 85, in _write
    self._instance.update_job_tick(self._tick)
  File "/opt/venv/lib/python3.8/site-packages/dagster/core/instance/__init__.py", line 1529, in update_job_tick
    return self._schedule_storage.update_job_tick(tick)
  File "/opt/venv/lib/python3.8/site-packages/dagster/core/storage/schedules/sql_schedule_storage.py", line 196, in update_job_tick
    with self.connect() as conn:
  File "/usr/local/lib/python3.8/contextlib.py", line 113, in __enter__
    return next(self.gen)
  File "/opt/venv/lib/python3.8/site-packages/dagster_postgres/utils.py", line 160, in create_pg_connection
    conn = retry_pg_connection_fn(engine.connect)
  File "/opt/venv/lib/python3.8/site-packages/dagster_postgres/utils.py", line 127, in retry_pg_connection_fn
    raise DagsterPostgresException("too many retries for DB connection") from exc

The above exception was the direct cause of the following exception:
sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not translate host name "docker_postgresql" to address: Temporary failure in name resolution

(Background on this error at: <https://sqlalche.me/e/14/e3q8>)

Stack Trace:
  File "/opt/venv/lib/python3.8/site-packages/dagster_postgres/utils.py", line 116, in retry_pg_connection_fn
    return fn()
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 3166, in connect
    return self._connection_cls(self, close_with_result=close_with_result)
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 96, in __init__
    else engine.raw_connection()
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 3245, in raw_connection
    return self._wrap_pool_connect(self.pool.connect, _connection)
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 3215, in _wrap_pool_connect
    Connection._handle_dbapi_exception_noconnection(
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 2069, in _handle_dbapi_exception_noconnection
    util.raise_(
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 3212, in _wrap_pool_connect
    return fn()
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/pool/base.py", line 307, in connect
    return _ConnectionFairy._checkout(self)
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/pool/base.py", line 767, in _checkout
    fairy = _ConnectionRecord.checkout(pool)
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/pool/base.py", line 425, in checkout
    rec = pool._do_get()
  File "/opt/venv/lib/python3.8/site-packages/sqlalchemy/pool/impl.py", line 256, in _do_get
    return self._create_connection()
  File "/opt/venv/lib/python3.8/site-packages/dagster/daemon/sensor.py", line 114, in __exit__
    self._write()
  File "/opt/venv/lib/python3.8/site-packages/dagster/daemon/sensor.py", line 85, in _write
    self._instance.update_job_tick(self._tick)
  File "/opt/venv/lib/python3.8/site-packages/dagster/core/instance/__init__.py", line 1529, in update_job_tick
    return self._schedule_storage.update_job_tick(tick)
  File "/opt/venv/lib/python3.8/site-packages/dagster/core/storage/schedules/sql_schedule_storage.py", line 196, in update_job_tick
    with self.connect() as conn:
  File "/usr/local/lib/python3.8/contextlib.py", line 113, in __enter__
    return next(self.gen)
  File "/opt/venv/lib/python3.8/site-packages/dagster_postgres/utils.py", line 160, in create_pg_connection
    conn = retry_pg_connection_fn(engine.connect)
  File "/opt/venv/lib/python3.8/site-packages/dagster_postgres/utils.py", line 127, in retry_pg_connection_fn
    raise DagsterPostgresException("too many retries for DB connection") from exc

Cameron Gallivan

09/15/2021, 12:22 AM

Any runs that we’re in progress freeze and will be stuck in the starting/started phase even after a server restart and need to manually be terminated. The deployed containers ignore

docker-compose down

and the command times out as well. This seems to only happen during sustained higher load times but I’m not sure if that is the sole/exclusive cause?

Devaraj Nadiger

09/15/2021, 7:26 AM

Does any body know solution for this?

Navneet Sajwan

09/15/2021, 10:14 AM

Hi, I am getting this error intermittently

Sebastian Napiorkowski

09/15/2021, 3:42 PM

I'm creating DynamicOutputs like this:

Copy code

@solid(input_defs=[InputDefinition(name="a_list")],
       output_defs=[DynamicOutputDefinition(dagster_type=dagster.Any)])
def emit_tasks_for_list(context, a_list: typing.List[typing.Any]):
    for num, item in enumerate(a_list):
        yield DynamicOutput(item, mapping_key=f'subtask_{num}')

it results in a weird grpc error, recently I've made the list general "[Any]", beforehand it was working with "[str]" any ideas:

Chris Evans

09/15/2021, 5:45 PM

What is the correct way to test a solid with a Nothing input arg via direct invocation?

Copy code

@solid
def start_after():
    return

@solid(required_resource_keys={"foo"}, input_defs=[InputDefinition("start_after", Nothing),])
def solid_requires_foo(context):
    return f"found {context.resources.foo}"

def test_solid_with_context():
    context = build_solid_context(resources={"foo": "bar"})
    assert solid_requires_foo(context, start_after=start_after()) == "found bar"

This results in a error:

Copy code

E       TypeError: solid_requires_foo() got an unexpected keyword argument 'start_after'

Arun Kumar

09/15/2021, 7:41 PM

Hi team, still facing this memory issue when the sensors are turned on. Running profilers within the sensor code did not provide any useful information. Is there any other way to profile the overall user code process?

Navneet Sajwan

09/15/2021, 8:09 PM

Hi everyone, How do I create multiple repos in one location using for loop? cc: @daniel

Chris Chan

09/15/2021, 9:17 PM

Is there a way to attach pre-configured loggers to a job? for example:

Copy code

log_configured = json_console_logger.configured({
    "log_level": "INFO"
})

my_job = my_graph.to_job(
    logger_defs={'my_logger': log_configured}
)

It seems like

my_logger

doesn’t do anything unless a config is explicitly provided in

to_job

Nazar Lysenko

09/16/2021, 7:47 AM

Hello! Does dagster make correct handling of SIGTERM, SIGINT signals? When we have gRPC service for repository with default launcher, running in a docker container, and we need to restart/redeploy container, it looks like service doesn't check if there are running pipelines. Could you suggest how to handle that?

Sebitien

09/16/2021, 9:18 AM

Hello, I use dynamic output to cut multiple calls to an external api into chunks that would not fit in memory if I was doing it directly into one chunk. One problem I am facing is that the map function in the map/collect logic requires to keep track of some state (not necessarily an integer). I came up with a kind of buffer logic using a resource in the map solid (see below) @solid(required_resource_keys={"buffer""},) def map_with_state(context, chunk): # fetch the state state = context.resources.buffer.get("state", default_value=0) # Computation result, new_state = computation(chunk, state) # Update the state for later computations context.resources.buffer.update(new_state) return result which seems to work so far. But since I will need to use that logic at multiple occasions, I was wondering if there was a more dagster-ly way to do it. Thanks !

Sebastian Napiorkowski

09/16/2021, 10:30 AM

Hi again, nobody helped me with my issue yesterday. I've re-created a small example, that just works:

Copy code

import dagster
from dagster import pipeline, solid, InputDefinition, DynamicOutputDefinition, DynamicOutput



@solid
def get_list():
    return [1, 2, "3", {"foo": "bar"}]


@solid(input_defs=[InputDefinition(name="a_list", dagster_type=dagster.List[dagster.Any])],
       output_defs=[DynamicOutputDefinition(dagster_type=dagster.Any)])
def emit_tasks_for_list(context, a_list):
    for num, item in enumerate(a_list):
        yield DynamicOutput(item, mapping_key=f'subtask_{num}')


@solid
def process_item(item):
    return str(item)


@solid
def concat_results(all):
    return " ".join(all)


@pipeline()
def dynamic_test():
    subtasks_result = emit_tasks_for_list(a_list=get_list()).map(process_item)
    concat_results(subtasks_result.collect())

But having the same code in a different context, ends in the gRPC error "Failure condition: Unexpected dynamic fan in dep created" any idea how to debug this? What does this code do: https://github.com/dagster-io/dagster/blob/8e95214331f94936dd9dbb8fc384979ed92fa34[…]2/python_modules/dagster/dagster/core/definitions/dependency.py

David Hyman

09/16/2021, 10:39 AM

👋 ~~any tips for logging? What am I doing wrong?~~ I had an extraneous

propagate=False

... oops! New logging feature works just fine; excellent, thank you 😁 (props to https://pypi.org/project/logging_tree/ for keeping me sane)

Sebastian Napiorkowski

09/16/2021, 12:33 PM

Hey guys, after a lot of debugging each part one by one the problem is actually very simple and it is definitely a bug: https://github.com/dagster-io/dagster/issues/4848

Edward Smith

09/16/2021, 3:51 PM

Hi All! First Time dagster user here. I'm porting an existing process to dagster, and using DynamicOutput to parallelize a bunch of sorting operations. Is there a way to have millions of DynamicOutputs without breaking the UI?

Jay Sharma

09/16/2021, 4:18 PM

Hi All, I have 10 pipelines that I want to run at the same time. What would be the best way to do this? Would I have to create a schedule that executes all the pipelines at the same time? I also want to see the progress bar of all pipelines running together? Is there a way to do that in the Dagit UI? Right now, I have to run each pipeline in a separate tab. I can only see one pipeline running at one time. Does Dagit allow us to see the progress for multiple pipeline at the same time? Essentially I want to view this screen for all the pipeline at once without switching tabs. Thanks!

Chris Chan

09/16/2021, 5:12 PM

By default, our Datadog integration forwards all stdout / stderr logs to Datadog. Is there a way to format / suppress the logging to stdout / stderr? In particular, all the DEBUG event logs make it hard to see useful information in Datadog, plus we’d like structured logs instead of plain text. The only way I’ve been able to change the Dagster event logging is to attach a json_console_logger named

console

to each pipeline and have a run config set the log level to INFO

David Hyman

09/16/2021, 5:45 PM

found a little logging bug with the new python functionality: https://github.com/dagster-io/dagster/issues/4853

Rohan Ahire

09/16/2021, 6:31 PM

Solids are eventually going to be replaced with the current experimental feature "Op". We are going to start a major project using dagster and were thinking to hold off and wait for the release of "Op". Looking for suggestion whether to hold or start development with solids? Similarly, pipeline will be replaced with "Job". Is just the naming convention being changed or something more and what about backward compatibility?

jay

09/16/2021, 7:12 PM

I am passing a

run_config

in to my pipeline as shown here: https://docs.dagster.io/concepts/configuration/config-schema It has following snipped of code as an example:

Copy code

execute_pipeline(
        config_example_pipeline,
        run_config={"solids": {"config_example_solid": {"config": {"iterations": 1}}}},
    )

Following is my

execute_pipeline

invocation:

Copy code

execute_pipeline(
	poc,
	run_config={"solids": {"with_run_config_alias": {"config": {"s": "testStr"}}}},
	)

Following is the pipeline:

Copy code

@pipeline
def poc():
	with_run_config_alias=with_run_config.alias("with_run_config_alias")
	with_run_config_alias()

Following is the solid executed by the pipeline above:

Copy code

@solid(config_schema={"s":str})
def with_run_config(context, s):
	print("with_run_config s: ", s)
	return to_str(s)

But I am getting this error:

Copy code

dagster.core.errors.DagsterInvalidConfigError: Error in config for pipeline poc
 Error 1: Missing required config entry "inputs" at path root:solids:with_run_config. Sample config for missing entry: {'inputs': {'s': '<selector>'}}

Yassine Marzougui

09/16/2021, 7:19 PM

Hi, Is it not possible to provide

solid_selection

in dagit playground?

Dan Stoner

09/16/2021, 7:50 PM

Looking for "how to organize my git repo" help. 'repository' of course being an overloaded concept here. I want to run

dagit

without any additional args so I have been trying to get

workspace.yaml

working but now I'm in what seems like Python import hell. I have an existing sample that works fine with

dagit -f hello_dagster.py

and all I want to do is move that into a subdirectory and have it available when running just

dagit

Francis Niu

09/17/2021, 2:50 AM

Hello Team, Can I use container as solid? e.g. each

docker run

command is a solid.

zafar mahmood

09/17/2021, 1:27 PM

Hi , i need a bit of guidance. I am testing solids (pytest) with context. now the solid takes the context and with arguments. Now during testing i get an error ' dagster.core.errors.DagsterInvalidInvocationError: Compute function of solid 'solid_function' has context argument, but no context was provided when invoking. ' an example which is exactly as referenced in documentation

Copy code

@solid( required_resource_keys = {'foo_str' , 'bar_int'})
def solid_function (context, some_json):
       var1 = context.resources.foo_str
       var2 = context.resources.bar_int
### now test 
def test_solid_function ():
    ### too build the context 
    from dagster import build_solid_context
    context = build_solid_context(resources={"foo_str":"some string ... ", 
                                             "bar_int": 344335 
     })
    solid_function(context=context , some_json= '{"name":"John", "age":30, "car":null}' )

can anyone share what i am doing wrong ? and getting the error ' DagsterInvalidInvocationError' Compute function of solid 'solid_function' has context argument, but no context was provided when invoking.

Deveshi

09/17/2021, 2:22 PM

Hi ! Is there an option to specify a database schema for run/event_log/schedule storage in dagster.yaml ? I am using the config for postgres_db. Thanks!

Eric Cheminot

09/17/2021, 3:28 PM

Hi! I'm trying to reproduce pipeline modes with new API. I understand from the documentation that the modes are (should?) be related to different Dagit instances, but I was using them with the same instance (and I think it makes sense to have those modes attached to the same instance), in my example to have different io_manager injected (to create side effects on different remote systems). So, now that the resources are attached to graphs to create jobs, what is the best option: just create 2 different jobs out of the same graph? Is there any other possibility? I understand the logic presented in the migration guide, I'm just not sure why it's enforced: am I relying on an anti-pattern of some sort? Thanks!

Yassine Marzougui

09/19/2021, 8:36 AM

Hi, When using memorization, is there a way to compute the input version at runtime? For example, when getting inputs from a mutable data store, I'd like the version to depend on the data itself in addition to the resource_version, solid_version, and the config_version. Something like this:

Copy code

@root_input_manager(
    version= dagster.hash_return_value() ???
)
def input_loader(_):
    return get_data_from_kv_store('key')

or with type loaders:

Copy code

@dagster_type_loader(
    loader_version= dagster.hash_return_value()
    ...
)
def foo_loader(context, config_value):
   get_data_from_kv_store(func(config_value))

or with solids:

Copy code

@solid(
    input_defs=[InputDefinition("foo", str), InputDefinition("foo", str)],
    version=[
        hash_solid_code(),
        dagster.hash_inputs() ???
    ]
)
def some_solid(context, foo, bar):
   ...

That way, in a memorized run, even if the config and the code are still the same, a solid would be executed again if one of its inputs changes. Is it maybe possible using `MemoizableIOManager`'s

has_output

function? I can't figure out how to do it.