Hi 👋 I am trying to load dbt assets from multiple projects. By reading https://dagster.slack.com/archives/C01U954MEER/p1669307372666059 I could define multiple assets using

dbt_resource_key

argument with

load_assets_from_dbt_projects

. But I am having problem include both dbt asset groups in the repository. I can include either one of them in the return statement of

repository

. But when I include both of them, I get below error.

dagster._core.errors.DagsterInvalidDefinitionError: Invalid dependencies: op "run_dbt_b7d19" does not have input "source_moo_moocards_moocards_moo_user". Available inputs: ['source_moo_recurly_billing_info', 'source_moo_recurly_invoices', 'source_moo_recurly_transactions']

The dbt assets are defined as below.

def get_dbt_group_name(node_info: Mapping[str, Any]):
    name = _get_node_group_name(node_info)
    return "dbt_" + name if name else None


# setup an `asset` with all dbt models
all_models = with_resources(
    # when loading assets from dbt, we invoke the `node_info_to_group_fn` argument
    # to label them all under a unified `group_name`
    load_assets_from_dbt_project(
        project_dir=DBT_PROJECT_DIR, profiles_dir=DBT_PROFILES_DIR, node_info_to_group_fn=get_dbt_group_name
    ),
    {
        "dbt": dbt_cli_resource.configured(
            {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR, "target": DBT_TARGET}
        )
    },
)

# setup an `asset` with dbt prep models
prep_models = with_resources(
    load_assets_from_dbt_project(
        project_dir=DBT_PREP_DIR,
        profiles_dir=DBT_PROFILES_DIR,
        dbt_resource_key="dbt_prep",
        node_info_to_group_fn=get_dbt_group_name,
    ),
    {
        "dbt_prep": dbt_cli_resource.configured(
            {"project_dir": DBT_PREP_DIR, "profiles_dir": DBT_PROFILES_DIR, "target": DBT_TARGET}
        )
    },
)

Could I get some insight on how to load both dbt assets into the repository? Thank you!

Hi. My use case is to use Dagster to generate a list of assets for each dataset in the platform we develop at Hugging Face (https://huggingface.co/datasets): list of ML “splits”, list of the first 100 rows, list of parquet files converted from the original dataset, list of column types, etc. The datasets are versioned with git, and we want to generate these assets for each commit of each dataset. I’m wondering if I should use the “software-based assets” paradigm for that, or rely on ops/graphs/jobs.

Hi, I have a question on the DockerRunLauncher. This may have been asked before but Slack doesn't show messages older than 90 days...Specifically, I use a host for Dagit and another for the user code server. If I use DockerRunLauncher to execute jobs, are the containers created on the Dagit host or the user code host? It wasn't very clear in the documentation https://docs.dagster.io/deployment/guides/docker I assume it is the user code server but just want to confirm. Thank you.

Hi Eric - would Kubernetes labels possibly help here? Those are very straightforward to set on runs currently and can help isolate to a specific job

Happy 2023! We ran into a weird ChildProcessCrashException with exit code -9 error for one job consistently. I am wondering if anyone has seen this before?

When trying to transition to the Definitons API with GRPC servers:

# old repository:
CMD ["dagster", "api", "grpc", "-h", "0.0.0.0", "-p", "4000", "-f", "FOO/repository.py", "--attribute", "myrepo"]
# ==> works fine

# new definitions:
CMD ["dagster", "api", "grpc", "-h", "0.0.0.0", "-p", "4000", "-m", "package_name"]
# ==> fails with: =>

DagsterInvariantViolationError: No repositories, jobs, pipelines, graphs, asset groups, or asset definitions found in "package_name"

CMD ["dagster", "api", "grpc", "-h", "0.0.0.0", "-p", "4000", "-m", "package_name.package_name"]
# ==> fails with: => No module named 'package_name/package_name'` while importing module package_name/package_name

Similarly to https://github.com/slopp/dagteam/blob/main/workspace.yaml: my workspace.yaml looks like:

load_from:
  - grpc_server:
      host: package_name
      port: 4000
      location_name: "package_name"

inside docker. But outside docker (during local development):

load_from:
  - python_package:
      package_name: package_name.package_name

Outside docker a

pip install -e .

has been executed - and the modules are resolvable. Inside docker, it would be convenient to not first volume-map/copy the file and then additionally pip install -e . it. It would be uesful (I think also for speedier reloading) to be able to pass a file like before: "-f", "FOO/repository.py". Is there any chance I can get this convenient behavior back when using definitions? My package (scaffolded from dagster cli as package_name/package_name/assets.py and package_name/package_name/__init__.py) contains in the init:

from . import assets
defs = Definitions(assets=(load_assets_from_modules([assets])))

However, I do not find any good way to specify the init file in the

-f

parameter. There, dagster also fails to resolve it due to missing modules.

Can we have multiple code locations in workspace.yaml? Documentation says we can but when I try to provide a python_package and python_file elements, dagit is not able keep the code server alive. The code server for package entry tries to start but is terminated subsequently. the python file entry never starts. When I try each option individually it works.

Hi all, what happens when you enable the following in the dagster helm chart:

"dagit": {
   "enableReadOnly": false, <---
   ...
}

While running the pipeline in dagster we are randomly getting the error below –

docker.errors.APIError: 500 Server Error for <http+docker://localhost/v1.41/images/create?tag=5.2.144&fromImage=><jfrog_url>%2F<project>%2Fdev%2F<container_image>: Internal Server Error ("Head https://<jfrog_url>/v2/<project>/dev/<container_image>/manifests/5.2.144: unknown: Authentication is required")
 File "/usr/local/lib/python3.8/dist-packages/dagster/core/instance/__init__.py", line 1732, in launch_run
   self._run_launcher.launch_run(LaunchRunContext(pipeline_run=run, workspace=workspace))
 File "/usr/local/lib/python3.8/dist-packages/dagster_docker/docker_run_launcher.py", line 152, in launch_run
   self._launch_container_with_command(run, docker_image, command)
 File "/usr/local/lib/python3.8/dist-packages/dagster_docker/docker_run_launcher.py", line 110, in _launch_container_with_command
   client.images.pull(docker_image)
 File "/usr/local/lib/python3.8/dist-packages/docker/models/images.py", line 465, in pull
   pull_log = self.client.api.pull(
 File "/usr/local/lib/python3.8/dist-packages/docker/api/image.py", line 429, in pull
   self._raise_for_status(response)
 File "/usr/local/lib/python3.8/dist-packages/docker/api/client.py", line 270, in _raise_for_status
   raise create_api_error_from_http_exception(e) from e
 File "/usr/local/lib/python3.8/dist-packages/docker/errors.py", line 39, in create_api_error_from_http_exception
   raise cls(e, response=response, explanation=explanation) from e
The above exception was caused by the following exception:
requests.exceptions.HTTPError: 500 Server Error: Internal Server Error for url: <http+docker://localhost/v1.41/images/create?tag=5.2.144&fromImage=><jfrog_url>%2Fdatamax%2Fdev%2F<container_image>
 File "/usr/local/lib/python3.8/dist-packages/docker/api/client.py", line 268, in _raise_for_status
   response.raise_for_status()
 File "/usr/local/lib/python3.8/dist-packages/requests/models.py", line 1021, in raise_for_status
   raise HTTPError(http_error_msg, response=self)

We are using two celery workers and rabbitmq. We are using the dagster.yaml file where we tried specifying the jfrog credentials using the following syntax -

registry:
      url: "<<https://registry.gitlab.com/v2>>"
      username: "myusername"
      password:
        env: DAGSTER_CONT_REGISTRY_DEPLOY_TOKEN

This issue is happening intermittently • We have tried restarting the docker services but in vain. • We tried to perform the docker login from inside the container and it worked but the pipeline was still failing intermittently. • We tried restarting all the containers, but the same issue was repeated. • We have a similar setup in another environment with the same container images and those are executing just fine. The docker version in the environment where it’s running and this environment where it's failing intermittently is the same.

I have created 3 python packages: - common utilities - team A: analytics - team B: ML These teams are all basic scaffolds and look similar to https://github.com/slopp/dagteam/blob/main/ml_project/ml_project_tests/test_project_loads.py (only the shared utilities package) is added. Problem: Either the classpath works fine for dagster - or for pytest. But not for both. I have a call to a function from the utilities package in team A:

from utilities.utilities import common_function

Dagster is only happy with this version above.

from utilities import common_function

Pytest only with this version here. XOR in either case one of thm (dasger, pytest throws a module not found error) or

ImportError: cannot import name 'common_function' from 'utilities' (unknown location)

.

NOTICE: all 3 packages have been installed with

pip install -e .

and should be natively callable.

Hi all! Can we customize job timeouts (default timeout and per job timeout)? Sometimes our dbt/extract jobs get stuck. And they might hang there executing for 20 hours until we see it and cancel them manually. We use DockerRunLauncher.

My guess is there’s a few resource initializations going on in the background, if I’m not mistaken they always get initialized before the green op bar. How are the startup times when you’re launching ops without resources?

Hi everyone! Is it possible to set up a type of OnStart hook which runs once exactly before an Op is executed?

Does someone know if the dagster-docker DockerRunLauncher is the same as Airflow

What is the proper way to do the following (Basically moving data from my databases to BQ with extra transformation): I have several assets

[A, B, C]

, and each gets data from different tables in a database and makes some transformations. The result of these assets is a dataframe that will be passed to another asset

X

responsible to send the data to BigQuery. I don't think this is the proper way to do that, I keep replicating the asset

X

because the table names are different. Is it possible to have a common asset that receives data from multiple assets and extra parameters (table name for example)? Also, should

X

be an asset?

The Dagster code server is shutting down and starting again every 20 - 30 seconds, is this okay or may be a problem?

Hi all I'm trying to mock out a context for an op I'm testing using

build_op_context

How do I specify tags in this function? Thanks!

When creating graph-backed assets, is it possible to only apply a partition_def to a subset of the resulting asset definitions? For example, say I have a graph that produces 3 assets - (1) needs to have daily partitioning, (1) needs weekly, and (1) does not need any partitioning. How would I go about this? Currently the only way I see to apply a partition is as follows:

graph_assets = AssetsDefinition.from_graph(the_graph, partitions_def=DailyPartitionsDefinition(...), can_subset=True)

You can instantiate Assets from existing ops and graphs, like

my_asset = AssetsDefinition.from_graph(my_graph, ...)

# alternatively
another_asset = AssetDefinition.from_op(my_op, ...)

Allowing you to re-use

my_graph

and

my_op

across different assets

Is it possible to create multiple job sensors?

Any clue on when this PR will be merged? https://github.com/dagster-io/dagster/pull/11057

Is there any documentation or examples of the GRPC server being extended to accomodate additional endpoints? MY company has containerized our code server, but we don't want to put another server in the same container.

Is there any way to access

op_config

within an IOManager? E.g.

@op(config_schema = {"uuid": str})
def op1(context) -> int:
    return 1

@op(config_schema = {"uuid": str})
def op2(context, a: int):
    return a * 2

@job(resource_defs={"io_manager": my_io_manager})
def my_job():
    op2(op1())

@io_manager
def my_io_manager():
    return MyIOManager()

class MyIOManager(IOManager):

    def handle_output(self, context, obj):
        # I want to access the uuid here to form the filepath to save the object

    def load_input(self, context):
        # I want to access the uuid here to form the filepath to load the object

Started to learn dagster from this great blog of @Pete Hunt and @schrockn it uses a lot of pip install commands in a script, it seems like the order is important. In this script pip throws a lot of resolution errors, but dagit seem to work fine anyways. Since pip dependency resolution is very simplistic and unreliable I’ve tried to do this tutorial with poetry, but then dagit won’t work due to some gql issues mentioned here before. I’ve managed to build the project with dagster 1.0.1, but in newer versions some dependency issues arise. Has somebody managed to have a simple “hello-world”-like dagster project with poetry and dagster 1.1.6+ ?

👋 Hey there. There are resources to launch jobs in databricks. I’m currently using the DatabricksPysparkStepLauncher, and it works quite well for our use cases. Have you checked out the integration? You may not be able to use this one for your autoloader, as it’s python only. However, there’s another job launcher you could use. AFAIK Autoloader is just a subset of spark structured streaming. However, it should also work. I havent used DBT much myself, and can’t really answer this part. dagster should also be able to orchestrate this — I assume you want to launch a databricks job that loads some delta file and writes it to postgres via JDBC?

I'm working on a POC for Dagster and I keep running into the following error:

% helm install dagster dagster/dagster               
Error: INSTALLATION FAILED: values don't meet the specifications of the schema(s) in the following chart(s):
dagster:
invalid character '<' looking for beginning of valuedagster-user-deployments:
invalid character '<' looking for beginning of value

I've tried using a local copy of

values.yaml

. I've also tried different versions of the helm chart, but I get the same error. My best guess is that it's reaching out and getting a bad http response. Does anybody know what's causing this?

Hey folks, I have a question re. dagster-daemon. Kindly redirect me to the relevant channel. To share with you a short context, we are running dagster pipelines (using python) on EC2 instance and have scheduled a couple of jobs every 2 hrs. Now the issue is that the jobs are not getting triggered as expected. Does anyone know what could be the root causes? For further clarification, please feel free to connect with me. P.S. As per the official documentation, “dagster-daemon” is responsible for the scheduler, backfill and sensors and we have “dagster-daemon” running on the EC2 instance. Thanks

Hello everyone, I want to check if the command is successful or not to continue processing the next code

Hi, I'm working on an op that processes a dataframe and make several API calls for each row. My initial implementation was to use DynamicOutput to manage the downstream ops. However, the APIs have rate limits and managing a few hundred thousand of parallel ops at one time seems quite expensive. Any other ideas on how I can approach this?

How can I provide a description to an asset created by

AssetDefinition.from_graph

? Given that the

@asset

decorator picks up docstring from the decorated function I expected the docstring of the graph to be “propagated” as the description of the created asset, but it doesn’t seem to be the case.