if I use a

get_dagster_logger()

log in a

sensor

evaluation, where would I find the log entries? eg

@sensor(job=my_job)
def my_job_sensor(context):
    log = get_dagster_logger()
    <http://log.info|log.info>("sensor initiated...")

Attempting to configure hooks to accomplish the following: • All ops trigger on failure. I believe this is set at the to_job(hooks={}) level with a graph. • A single op triggers on success (Unclear how to accomplish this with a graph) The documentation for the per op only shows jobs. Graph doesn’t accept resource_defs. Setting the resource def at the job level and trying to use it at the op doesn’t work. Trying to set the resource at the op level and only reference the hook doesn’t work. I’m at a bit of a loss on how to make this work with a graph… Sample implementation

# Create a graph
@graph
Def pipeline(): 
	Output = my_ops()
    my_ops_result.with_hooks({slack_message_on_success, slack_connection})()
 
# Convert to job
 run = pipeline.to_job(resource_defs={'aws_session': aws_session,
                                      'db_connection': db_connection,
                                      'slack_connection': slack_connection
                                      },
                       hooks={slack_message_on_failure},
                       name='pipeline_{0}'.format(env),
                       config=config
                       )

Result: DagsterInvariantViolationError

Hey! I'm new to using dagster and I was wondering if it was possible to set an environment variable before any jobs execute so that a docker entrypoint can retreive those env variables and use them for a script.

Hey Dagsters, I'm trying to supply some runtime config for an IOManager. I would like to supply a build number so that all assets materialised by the job with be written to S3PATH + Build Number + Step Key. I'm defining the config like so:

@io_manager(required_resource_keys={"pyspark"},
            config_schema={"path_prefix": str, "stage_prefix": str, "build_number": str})
def parquet_io_manager():
    return ParquetIOManager()

Creating the resource:

@repository
def repository():
    return [
        with_resources(
            all_assets,
                { "bronze_parquet_io_manager": parquet_io_manager.configured({"path_prefix": "<s3://817187305846-dagster-test>", "stage_prefix": "bronze"}) }
        ),

Note - I'm omitting some other resource config for brevity but intentionally also omitting the "build_number" config as I want this to be populated at runtime. Then supplying the following at launch via dagit:

resources:
   bronze_parquet_io_manager:
    config:
      build-number: "build-600"

Is supplying config to IOManagers like this supported?

This message was deleted.

Hi, Im currently running the dagster-daemon run but it show that the logs file existed. So • Does i have to remove the

logs

folder for every runtime?

Hi everyone, would anyone know why my dagster-daemon scheduler appears to not be running when I use docker, thats the current code in my dagster.yaml file and my docker-compose

it says the sensor daemon is running but not the scheduler

Answer to my own question, was following the guide and adding the postgres schedule storage

in case anyone runs into this in the future

Hi Dagsters, I have a real-time pipeline use Redis as database for storing and fetching data, now I have tried to port it to a offline pipeline for day-part backfill by using Dasgter. After search across the document, all I need is a dynamic resource to create Redis container and then tear down after a mapped ops finished. But it seems the resource in Dagster stops after entire jobs finished. Then I tried sensor to monitor the ops, first I will create Redis for each mapped ops, and my idea is to shutdown the Redis container once the job success or fail. That would work but it seems a little bit weird, anyone has a better way to structure the pipeline?

How might I construct a graph that does something like this: (I know this doesn't work, but I think/hope my intent is clear enough to see what I'd like to happen). Looking to configure an

op

based on the output of a different

op

from dagster import op, graph
from dagster_k8s import k8s_job_op

@op
def get_greeting():
  return "hello"

@graph
def my_graph():
   greeting = get_greeting()

   greeter = k8s_job_op.configured(
     {
        "image": "busybox",
        "command": ["/bin/sh", "-c"],
        "args": ["echo " + greeting], #obv this doesn't work, but not sure what to do
        "load_incluster_config": False,
        "namespace": "default"
      },
      name="say_greeting"
   )
   greeter()

Hi! It's been a bit. I'm deploying a new ETL, and I'm running into an error running in dev vs running locally with dagit. I also tried it downloading the prev output that goes into the failing op from the gcs bucket, and running that op in a notebook, and it actually works. I checked the package versions in the container running it and my local env, and the dagster, dagster-cloud, pandas, and numpy versions are all the same. Does anyone have another lead?

TypeError: int() argument must be a string, a bytes-like object or a number, not 'NAType'
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/execution/plan/utils.py", line 47, in solid_execution_error_boundary
    yield
  File "/usr/local/lib/python3.9/site-packages/dagster/_utils/__init__.py", line 430, in iterate_with_context
    next_output = next(iterator)
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/execution/plan/compute_generator.py", line 74, in _coerce_solid_compute_fn_to_iterator
    result = fn(context, **kwargs) if context_arg_provided else fn(**kwargs)
  File "/expo-analytics/secmaster/lipper/assets/fund_updates.py", line 31, in new_and_existing_fund_instruments
    existing_funds = bq_lipper.check_if_fund_in_yb_instrument_universe(
  File "/expo-analytics/secmaster/lipper/resources/bigquery_lipper.py", line 392, in check_if_fund_in_yb_instrument_universe
    all_matches_perm_id_yb_id = pd.merge(
  File "/usr/local/lib/python3.9/site-packages/pandas/core/reshape/merge.py", line 121, in merge
    return op.get_result()
  File "/usr/local/lib/python3.9/site-packages/pandas/core/reshape/merge.py", line 715, in get_result
    join_index, left_indexer, right_indexer = self._get_join_info()
  File "/usr/local/lib/python3.9/site-packages/pandas/core/reshape/merge.py", line 966, in _get_join_info
    (left_indexer, right_indexer) = self._get_join_indexers()
  File "/usr/local/lib/python3.9/site-packages/pandas/core/reshape/merge.py", line 940, in _get_join_indexers
    return get_join_indexers(
  File "/usr/local/lib/python3.9/site-packages/pandas/core/reshape/merge.py", line 1481, in get_join_indexers
    zipped = zip(*mapped)
  File "/usr/local/lib/python3.9/site-packages/pandas/core/reshape/merge.py", line 1478, in <genexpr>
    _factorize_keys(left_keys[n], right_keys[n], sort=sort, how=how)
  File "/usr/local/lib/python3.9/site-packages/pandas/core/reshape/merge.py", line 2165, in _factorize_keys
    rk = ensure_int64(np.asarray(rk))
  File "pandas/_libs/algos_common_helper.pxi", line 93, in pandas._libs.algos.ensure_int64

hey all 🙂 I am trying to write a dagster project that will consume tables from a url and then do some postprocessing in dbt. the dagster side of things will a) download

csv.gz

files from the net to a bucket, b) run a create external table sql statement to register the table. Then DBT takes over and consumes the csv.gz files and partions/buckets and transforms to parquet. with my current setup I have all the raw tables defined as dbt sources and do a simple

select * from table

with some extra config to handle bucketing/partitioning/parqueting. Unforunately when I try to load the project in dagster now DBT is looking at the sources, attempting to retrieve schema and then failing during repo init. Is there a pattern I could use here that could help here? failure pictured

Hi Dagsterinos, Dagster newbie here. I’m having an issue where I’d like a monthly partitioned asset to be run on a daily schedule. I was under the impression this was possible, but its a bit unclear from the docs how I would do it. Is there a simple example in the docs for this? If I create a schedule using

build_schedule_from_partitioned_job

, it is a monthly schedule because it’s a monthly partition. When I try to do something like:

my_month_partitioned_job = define_asset_job(name="job_name", selection="my_monthly_partitioned_asset", partitions_def=monthly_partition_def)
my_schedule = ScheduleDefinition(job=my_month_partitioned_job, cron_schedule="21 16 * * *")

I get an error like:

dagster._check.CheckError: Invariant failed. Description: Tried to access partition_key for a non-partitioned run

If I leave out the job partition definition:

my_month_partitioned_job = define_asset_job(name="job_name", selection="my_monthly_partitioned_asset")
my_schedule = ScheduleDefinition(job=my_month_partitioned_job, cron_schedule="21 16 * * *")

I get the same error as above:

dagster._check.CheckError: Invariant failed. Description: Tried to access partition_key for a non-partitioned run

Part of whats going on might be that my monthly partitioned asset is accessing the context to do what it needs to do:

@asset(partitions_def=monthly_partition_def)
def my_monthly_partitioned_asset(context):

    partition_key = context.asset_partition_key_for_output()
    ... use partition key to understand what month partition it is...
    return stuff

stack trace looks like this:

Stack Trace:
  File "lib/python3.9/site-packages/dagster/_core/execution/plan/utils.py", line 47, in solid_execution_error_boundary
    yield
  File "lib/python3.9/site-packages/dagster/_utils/__init__.py", line 430, in iterate_with_context
    next_output = next(iterator)
  File "lib/python3.9/site-packages/dagster/_core/execution/plan/compute_generator.py", line 73, in _coerce_solid_compute_fn_to_iterator
    result = fn(context, **kwargs) if context_arg_provided else fn(**kwargs)
  File "assets.py", line 39, in my_monthly_partitioned_asset
    <http://context.log.info|context.log.info>(context.asset_partition_key_for_output())
  File "lib/python3.9/site-packages/dagster/_core/execution/context/compute.py", line 370, in asset_partition_key_for_output
    return self._step_execution_context.asset_partition_key_for_output(output_name)
  File "lib/python3.9/site-packages/dagster/_core/execution/context/system.py", line 811, in asset_partition_key_for_output
    start, end = self.asset_partition_key_range_for_output(output_name)
  File "lib/python3.9/site-packages/dagster/_core/execution/context/system.py", line 806, in asset_partition_key_range_for_output
    return PartitionKeyRange(self.partition_key, self.partition_key)
  File "lib/python3.9/site-packages/dagster/_core/execution/context/system.py", line 334, in partition_key
    check.invariant(
  File "lib/python3.9/site-packages/dagster/_check/__init__.py", line 1470, in invariant
    raise CheckError(f"Invariant failed. Description: {desc}")

Hey All, also a dagster newbie. I have a question on the deployment side. We have Dagster deployed into a GKE Cluster and I was trying to get a run worker pod to deploy alongside an initContainer, and I’m not 100% clear on where I would go to define an initContainer for a worker. Some googling and doc review pointed me towards this page: https://docs.dagster.io/deployment/guides/kubernetes/customizing-your-deployment#separately-deploying-dagster-infrastructure-and-user-code But from that page I’m not clear on how I would modify the user code customization to include an initContainer when a worker pod is spawned. Would it be here: https://github.com/dagster-io/dagster/blob/master/helm/dagster/charts/dagster-user-deployments/templates/deployment-user.yaml#L35?

Hi, is it possible to materialise an asset with a run config and partition key. This is the error I currently get using materialise_to_memory "Cannot provide both run_config and partition_key arguments to `execute_in_process`"

Having some issues materializing Airbyte assets as described by this documentation. All indications is that it is able to connect to my AIRBYTE_HOST, but perhaps I’m passing the wrong value? Since Airbyte is deployed to the same cluster as my Dagster app, instead of

localhost

as the host, I’m using

airbyte-airbyte-server-svc

It definitely is able to connect, but doesn’t seem to be pulling assets.

Hi all! Is it possible to have an asset with different

key

and "name" values? I want to use the

key

to refer to the asset programatically (for example, in yielding `AssetMaterialization`s) and the "name" to be a nice, meaningful name to be displayed on the UI

I am getting an error with parmiko indicating that Blowfish has been deprecated. This issue indicates that the issue has been fixed: https://github.com/paramiko/paramiko/issues/2038 I've never contributed to open source but I think the fix would go here: https://github.com/dagster-io/dagster/blob/dc4fe9b3f7b306da959c30eb99906662c5cb3d7e/python_modules/libraries/dagster-ssh/setup.py Happy to contribute or do research but not sure how to start.

hey all, how can we added non argument deps for assets produced using

AssetsDefinition.from_graph

?

Hi all! I’m looking for good design patterns when ingesting data from several files in the same S3-bucket into a data warehouse. The only thing I need to do in my Software-Defined Asset is loading the file into a pandas DataFrame and letting my IOManager do the rest. It feels weird and against all manner of DRY-principles to code 20 functions identical in all but their names. Would it be possible to design a SDA factory and instantiate them through job/sensor definitions?

Hi team! I continue to investigate Dagster performance in case of DynamicOutput. Here are some observations: • I modeled a simple job that does something like this:

dynamic_int_sequence().map(twice).collect()

that returns list like

[2,4,6,...]

• I use

mem_io_manager

and

in_process_executor

, running on my local machine (enough memory, good cpu and fast ssd) • for 1000 elements pipeline takes ~*100 sec* to complete, generating 10k events •

dynamic_int_sequence

which just returns 1000 consecutive DynamicOutput integers takes 12 sec This demonstrates huge overhead on relatively small amount of simple operations. So I have a question: is it convenient to use DynamicOutput in such cases?

Hello! Are there any suggested/sane limits for the number of defined entities in Dagster - e.g. max number of Repositories, Assets/Ops or Jobs?

I would like to link directly to the logs of a job from a failure hook. Can I access the run_id of a job from within an op?

Hello. Do dagster logger can catch logs from logger of other process? For example.: Op A starts process with subprocess.Popen. This process has configured logger. I want to be able to display logs from subprocess using dagster logger. Is it possible? What do I need to do?

Hi! If I have an assets A depending on, say, two upstream assets B and C, does Dagster execute

IOManager.load_input

of B and C sequentially or in parallel? Also, a more specific use-case: If I have Assets A and B which are both subset of rows in a database table, how can I setup an

IOManager.load_input

to load both using the same query and then splitting the results and redirecting then to each Asset?

Hello dagsters! I'm trying this 2 things: 1. Have an asset w/o partition downstream of an asset w/ partition. This second asset would

aggregate

the partitioned asset 2. Have an asset w/ partition downstream of an asset w/ partition but would like to receive the data

rolling

, wouldn't map 1 to 1 but rolling n to 1 I believe this could be accomplished by PartitionDefinition but having trouble finding how.

Hello Dagsters I was trying dagster with aws, would like your thoughts on 1) Any idea whther we can call glue job with op or job? 2) Any help/pointers on how to add mysql RDS as source Thank you & have a great day !

Hi I downloaded the github example with pyspark, but it doesn't worked. Here is the error message: