Hello ! We're using dagster_pandas and we are stuck with some test cases: how to create fixtures/mocks of a specific dagster_mandas Data Frame? (that is from a pd.DataFrame)

Hi, I’m trying out Dagster for the first time and was trying to get the EMR PySpark example running. https://github.com/dagster-io/dagster/tree/0.13.14/examples/emr_pyspark I’m running on MacOS 12.1 (Apple M1) with Python 3.9.7 and Microsoft OpenJDK 11. When I try to do a local run in Dagit, I get the following error message:

dagster.core.errors.DagsterExecutionHandleOutputError: Error occurred while handling output "result" of step "make_people":
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/execution/plan/execute_plan.py", line 222, in _dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/execution/plan/execute_step.py", line 328, in core_dagster_event_sequence_for_step
    for evt in _type_check_and_store_output(step_context, user_event, input_lineage):
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/execution/plan/execute_step.py", line 383, in _type_check_and_store_output
    for evt in _store_output(step_context, step_output_handle, output, input_lineage):
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/execution/plan/execute_step.py", line 512, in _store_output
    for elt in iterate_with_context(
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/utils/__init__.py", line 389, in iterate_with_context
    return
  File "/opt/homebrew/Cellar/python@3.9/3.9.10/Frameworks/Python.framework/Versions/3.9/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/execution/plan/utils.py", line 73, in solid_execution_error_boundary
    raise error_cls(
The above exception was caused by the following exception:
TypeError: cannot pickle '_thread.RLock' object
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/execution/plan/utils.py", line 47, in solid_execution_error_boundary
    yield
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/utils/__init__.py", line 387, in iterate_with_context
    next_output = next(iterator)
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/execution/plan/execute_step.py", line 504, in _gen_fn
    gen_output = output_manager.handle_output(output_context, output.value)
  File "/Users/miles/.local/share/virtualenvs/dagster-KC6aDfP0/lib/python3.9/site-packages/dagster/core/storage/fs_io_manager.py", line 122, in handle_output
    pickle.dump(obj, write_obj, PICKLE_PROTOCOL)

Any ideas of what’s going wrong? The failing function is making a data frame from static data, so I’m struggling to see where the problem is.

Hi all, I am in the process of upgrading our code base from the old dagster api to use jobs. I can see that the JobDefinition class takes ModeDefinition and PresetDefinition’s as inputs. It would make it a lot easier for transitioning our current pipelines with modes and presets into jobs using the JobDefinition class with graph, mode and preset inputs. My question is, will the ModeDefinition and PresetDefinition classes be deprecated in the future, or will they remain but just as internal apis (as they are not referred to in the documentation anymore)?

Is the only way to get asset materializations within a sensor for a given run_id by creating a cursor for specific runs?

cursor_dict = json.loads(context.cursor) if context.cursor else {}

asset_key = context.dagster_run.run_id
asset_key_cursor = cursor_dict.get(asset_key)

cursor_events_map = dict()

asset_key_event_records = context.instance.get_event_records(
    EventRecordsFilter(
        event_type=DagsterEventType.ASSET_MATERIALIZATION,
        asset_key=AssetKey(asset_key),
        after_cursor=asset_key_cursor,
    ),
)

cursor_events_map[asset_key] = asset_key_event_records

# makes sure there are events for all cursor keys
if not all([bool(v) for v in cursor_events_map.values()]):
            return

cursor_update = dict()
for cursor_key, cursor_events in cursor_events_map.items():
    cursor_update[cursor_key] = cursor_events[0].storage_id

context.update_cursor(json.dumps(cursor_update))

Has anyone tried running pyspark on EMR on EKS? It uses “EMR Containers”, which has a different API: https://docs.aws.amazon.com/emr-on-eks/latest/APIReference/API_StartJobRun.html I’m hoping that a “job run” is analogous to “steps”, but would love to hear from other people who might have already tried.

Hello, I am upgrading old dagster code from 0.11.13 to current 0.13.14. There is a deprecated function called "IContainSolids" that I need to replace since it no longer in the code. What would be the best option to replace this for current version 0.13.14? Here is the function definition before it was removed. https://github.com/dagster-io/dagster/blob/cd9826793481d631225f0ac188f03ee9ca293ac[…]hon_modules/dagster/dagster/core/definitions/solid_container.py

I posted before Christmas but haven't had any reply, so this is an abridged version of my problem. I have a job that needs to execute on weekdays only. I wasn't able to do this with a

daily_partitioned_config

and had to resort to using a

dynamic_partitioned_config

. When the job is run via this it conveniently leaves a

partition

tag against the run. But normally the job is launched via a schedule. Using the

@schedule

decorator I can specify a cron expression that supports my weekday requirement. The schedule looks something like this, where

daily_run_config

is a convenience function that provides the run_config I need:

@schedule(cron_schedule="0 3 * * 1-5", execution_timezone='Australia/Melbourne', job=Jobs.daily_positions_upload)
def upload_schedule(context):

    effective_date = (context.scheduled_execution_time - BDay(2)).strftime("%Y-%m-%d")
    config = daily_run_config(effective_date)
    return config

However, I would prefer for the schedule to use the

dynamic_partition_config

I've already got. This would leave the

partition

tag against the run, whereas the current scheduled run doesn't. But I can't return a

dynamic_partitioned_config

from a schedule decorated function - I get this error:

dagster.check.CheckError: Object DynamicPartitionsDefinition(partition_fn=<function get_effective_dates1 at 0x000001F13297B798>) is not a TimeWindowPartitionsDefinition. Got DynamicPartitionsDefinition

Thanks Paul

Does any have experience with running Dagster in Digitalocean’s app platform? It would possible since Dagster is just a python app. dagster-daemon can run as a worker. A possible problem is that there is no way of password protecting dagit?

Hello, I have an architecture question. I am running dagster on prem infrastructure and I want to call (REST API) an external service that will execute a ops (or few). This external service takes minutes to run and give back results. How would you implement this to manage this coupling ? Ask me details if it is too concise 🙂

Do people use dagster-dbt to manage the relationship between a target and staging table?

Is anyone else experiencing long pauses between ops when running on docker in M1 macs? On intel it seems way faster… (using multi process executor, and code is mounted as a volume)

I've created a resource database connection. But whenever I want to use it I get the error

required_resource_keys is not one of ['frozenset' , 'set']. got ['myconnection'] which is type <class 'list'>

What I'd like to know is: What does this mean, how do I fix this, and what is best practice? (and I'd like to see an example in the docs?)

Hey all general question here - In what instance would I choose to use celery k8s runner vs k8s runner? There seems to be much more infra overhead to it

Hey All, just wanted some clarification since I couldn't find any in the docs for this - what is the difference in pipelineRun section of the helm chart vs the user-code-deployment chart? I always thought the image which was hosting the grpc server would also let the daemon know which image to execute with

My op creates a pandas dataframe, I don't want to validate it so I don't think I need dagster-pandas? My op function has the type hint

-> pd.DataFrame:

But when I'm running a test I see the error

Type check failed for op "nameofop" output "result" -expected type "DataFrame". Description Value of type<class function>. ... expected value to be of Python type pandas.core.frame.DataFrame

I thought it was when I return the dataframe?

Hey folks. I'm working through the migration to jobs/ops/graphs and one of the last pieces is to move my usage of

PresetDefinition

. The examples show how to handle that in the case that you have one being passed to

preset_defs

in the

@pipeline

decorator, but not a good example of handling a list of them. It seems that the answer is that I have to create multiple instances of the job to be able to pass it into the schedule with the appropriate config?

If I define a resource for a public web endpoint I want to hit with 5 parallel requests at a time, (e.g.

MyResource.fetch(…)

), would it be reasonable if I reference the resource from within a threadpool, or would there be known issues with that kind of behavior?

Is there a way to define an OpDefinition that functions like a .collect for graphs with dynamic outs? I'm creating everything via config-based factory so can't use the .map/.collect from the documentation.

is there a way to access a resource by a variable defined somewehre else?

RESOURCE_KEY = "some_key"

context.resources[RESOURCE_KEY]

Hi all, when running multiple workspaces in a single dagit instance, we can filter repositories within those workspaces - meaning we can effectively hide jobs in other repositories. However, the runs for these 'foreign' jobs still show in the 'Runs' menu and I can't work out a simple way to hide them. P.S. I realise it's considered bad practice to have multiple workspaces per Dagit instance but we're some way off getting our dev ops tooling to support multiple dagits.

Hi, I am trying the Dagster with dbt integration. I have a config file where i am passing the project_dir and the profile_dir that is required for the DBT models to run. When I try to run the DBT models i keep getting the error as dagster.core.errors.DagsterExecutionStepExecutionError: Error occurred while executing op "op_run_dbt": and the associated exception is The above exception was caused by the following exception: AttributeError: 'function' object has no attribute 'run' I am confused - How do i convert a resource definition to a resource ? Example - dagster.core.definitions.resource_definition.ResourceDefinition (This is the object type I am getting in Dagster UI) to dagster_dbt.cli.resources.DbtCliResource' (This is the object type that I require to run models in DBT) Please let me know if any additional information is required

Hey folks. I'm trying to use the same op twice in a job, but passing different configuration to it each time. I'm quite new to the whole config_schema stuff. Is this possible? Code below (obviously Dagster doesn't like the fact I'm using strings as input variables right now):

@op
def push_to_kafka(parsed_feed, topic):
    producer = KafkaProducer(bootstrap_servers='XXXXXXX',
                             key_serializer=lambda x: x.encode('utf8'),
                             value_serializer=lambda x: json.dumps(x).encode('utf8'))
    for entry in parsed_feed:
        producer.send(topic,
                      key=entry['id'],
                      value=entry)
    get_dagster_logger().info(f"Successfully pushed to Kafka {topic}")


@job
def download_hacker_news_job():
    metadata = parse_rss_feed(download_rss_feed())
    push_to_kafka(metadata, 'news-metadata')
    push_to_kafka(download_article_content(metadata), 'news-content')

I'm receiving the error.

dagster.core.errors.DagsterInvariantViolationError: research_and_development not found at module scope in file repo.py.

Only when my pipeline is triggered by a schedule. It runs fine otherwise. I have absolutely no idea what code change has caused this, it's been working smoothly for months (I am still assuming I've broken something with a code change of course).

Hi All, I'm executing pipeline using python execute_pipeline api and for that I'm generating a pipeline using PipelineDefinition. Now when I try to execute the pipeline in multiprocess mode it is not possible to reconstruct a PipelineDefinition object. Now how I can make PipelineDefinition as reconstructable in order to execute it in execute_pipeline using python.

Dagster docs going 404 for anyone else?

Is it possible to have an op that takes one regular input and one dynamic input? Almost like a curried op.

Are there any limitations on the number/length of tags a run can have? I recently started batching tasks into a single job and previously, when each job was a single task, I’d tag it with a name for searching in dagit. Now that the tasks are batched i’d like to achieve something similar and didn’t know if either of these approaches had any obvious problems: 1. Each job gets a dynamic number of tags, one for each task:

tags={
    'task1': 'taskname1',
    'task2': 'taskname2',
    'task3': 'taskname3',
    'task4': 'taskname4',
    ...
    'task50': 'taskname50',
}

2. Or concatenate the tasknames into a single string:

tags={
    'tasks': 'taskname1,taskname2,taskname3,taskname4,...,taskname50'
}

Hey dagster team 🙂 ! If I want my user-deployment grpc and other infra to be deployed in a less resource intensive node pool, but wanted all the k8s pods to execute in another pool, where would I configure that in the helm chart?

Hi All, I just received the below unexpected error when trying to look at the

read_files

graph in the attached image. Would anyone be able to help me figure out what's causing the error?

Operation name: PipelineExplorerRootQuery

Message: '_TypeHintInferred[pandas.core.frame.DataFrame]'

Path: ["pipelineSnapshotOrError","solidHandles",0,"solid","definition","outputDefinitions",0,"type"]

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

Stack Trace:
  File "C:\ProgramData\Miniconda3\envs\spence\lib\site-packages\graphql\execution\executor.py", line 452, in resolve_or_error
    return executor.execute(resolve_fn, source, info, **args)
  File "C:\ProgramData\Miniconda3\envs\spence\lib\site-packages\graphql\execution\executors\sync.py", line 16, in execute
    return fn(*args, **kwargs)
  File "C:\ProgramData\Miniconda3\envs\spence\lib\site-packages\dagster_graphql\schema\solids.py", line 79, in resolve_type
    return to_dagster_type(
  File "C:\ProgramData\Miniconda3\envs\spence\lib\site-packages\dagster_graphql\schema\dagster_types.py", line 25, in to_dagster_type
    dagster_type_meta = pipeline_snapshot.dagster_type_namespace_snapshot.get_dagster_type_snap(
  File "C:\ProgramData\Miniconda3\envs\spence\lib\site-packages\dagster\core\snap\dagster_types.py", line 48, in get_dagster_type_snap
    return self.all_dagster_type_snaps_by_key[key]

Hei! What is the best way to create jobs by both time and non time attribute? Lets say I have a job which I need to run for N countries every day. I can solve this by having a schedule which yields runrequests for each country for each day. But this doesn't work for for backfills - if I have partition by country I can't create a time partition on it. I could handle it inside the job - just loop over the countries but Ideally I would want to launch job per country/period separately. Guess what I am asking is how do i do time backfills for static_partitioned_config jobs?