Seeing the below error with my dagster and R project. This project was largely developed on my work machine, which is running Windows 10. Unfortunately, I'm seeing the below error when trying to materialize the

iris_r

asset from my personal machine, which is running Ubuntu 22.04: dagster._core.errors.DagsterInvariantViolationError: op 'iris_r' did not yield or return expected outputs {'iris_r_species_name_check_r', 'iris_r_no_missing_sepal_width_check_r', 'iris_r_no_missing_petal_length_check_r', 'iris_r_no_missing_petal_width_check_r', 'iris_r_no_missing_sepal_length_check_r'}. Did you forget to

yield from pipes_session.get_results()

or

return <PipesClient>.run(...).get_results

? If using

open_pipes_session

,

pipes_session.get_results

should be called once after the

open_pipes_session

block has exited to yield any remaining buffered results via

<PipesSession>.get_results()

. If using

<PipesClient>.run

, you should always return

<PipesClient>.run(...).get_results()

or

<PipesClient>.run(...).get_materialize_result()

. File "/home/porlando/.cache/pypoetry/virtualenvs/dagster-and-r-kS5e8P_l-py3.10/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_plan.py", line 286, in dagster_event_sequence_for_step for step_event in check.generator(step_events): File "/home/porlando/.cache/pypoetry/virtualenvs/dagster-and-r-kS5e8P_l-py3.10/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 478, in core_dagster_event_sequence_for_step for user_event in _step_output_error_checked_user_event_sequence( File "/home/porlando/.cache/pypoetry/virtualenvs/dagster-and-r-kS5e8P_l-py3.10/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 160, in _step_output_error_checked_user_event_sequence for user_event in user_event_sequence: File "/home/porlando/.cache/pypoetry/virtualenvs/dagster-and-r-kS5e8P_l-py3.10/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 95, in _process_asset_results_to_events for user_event in user_event_sequence: File "/home/porlando/.cache/pypoetry/virtualenvs/dagster-and-r-kS5e8P_l-py3.10/lib/python3.10/site-packages/dagster/_core/execution/plan/compute.py", line 260, in execute_core_compute raise DagsterInvariantViolationError(message) The source code is available here: https://github.com/philiporlando/dagster-and-r Let me know if you're unable to browse that repo, and I'll paste the code in a new message.

The error message you're encountering indicates that the operation (op) named

iris_r

is expected to yield or return certain outputs, but it did not. This is a common issue when using Dagster's Pipes functionality, which allows for communication between Dagster and external processes. The error message suggests that you might have forgotten to use

yield from pipes_session.get_results()

or

return <PipesClient>.run(...).get_results()

in your code. These are necessary to ensure that the results from an external process are properly communicated back to Dagster. Here are a few steps you can take to troubleshoot this issue: 1. Check the Implementation: Ensure that your

iris_r

operation is correctly yielding or returning the expected outputs. If you're using a

PipesSession

, make sure to call

pipes_session.get_results()

after the

open_pipes_session

block has exited. If you're using a

PipesClient

, ensure that you return the results using

PipesClient.run(...).get_results()

or

PipesClient.run(...).get_materialize_result()

. 2. Review the Code: Since the code is running on a different operating system (Ubuntu 22.04) than it was developed on (Windows 10), there might be OS-specific issues at play. Review the code to ensure there are no such dependencies that could cause the code to behave differently across operating systems. 3. Check for Version Differences: Ensure that the versions of Dagster and any related libraries are consistent between your work machine and your personal machine. Differences in versions can sometimes lead to unexpected behavior. 4. Inspect the Logs: The stack trace provided in the error message can give you clues about where in the code the issue is occurring. Follow the stack trace to pinpoint the exact location in your code where the outputs are not being yielded or returned as expected. 5. Access the Repository: If you need more specific guidance, accessing the source code repository would be helpful. However, as a Dagster AI BOT, I do not have the capability to browse external repositories. If you can share the relevant parts of the code here, I can provide more targeted advice. Remember to ensure that any sensitive information is removed before sharing your code publicly. If you're unable to resolve the issue with these steps, please provide the relevant code snippets, and I'll do my best to assist you further.

@Scout, here is the iris.R script. It relies on reticulate to interact with the dagster pipes functions. It contains a mix of business logic written in R, and reticulate/Python/Dagster boilerplate. library(reticulate) library(readr) library(glue) library(magrittr) # Ensure that RETICULATE_PYTHON is set to the poetry path reticulate::py_config() stopifnot(reticulate::py_module_available("dagster_pipes")) # Import Python modules # R doesn't support selective imports like Python, so you have to do this # to avoid typing the full namespace path repeatedly... os <- reticulate::import("os") dagster_pipes <- reticulate::import("dagster_pipes") open_dagster_pipes <- dagster_pipes$open_dagster_pipes PipesContext <- dagster_pipes$PipesContext # https://rstudio.github.io/reticulate/articles/calling_python.html#with-contexts with(open_dagster_pipes() %as% pipes, { context <- PipesContext$get() context$log$info("This is a log from R!") data(iris) context$log$info(head(iris)) context$log$info(os$environ["MY_ENV_VAR_IN_SUBPROCESS"]) output_dir <- Sys.getenv("OUTPUT_DIR") context$log$info(glue::glue("output_dir: {output_dir}")) output_file <- glue::glue("{output_dir}/iris.csv") context$log$info(output_file) context$report_asset_materialization() # Ensure that Sepal.Length field does not contain any NAs context$report_asset_check( asset_key = "iris_r", passed = reticulate::r_to_py(all(!is.na(iris$Sepal.Length))), check_name = "no_missing_sepal_length_check_r", ) # Ensure that Sepal.Width field does not contain any NAs context$report_asset_check( asset_key = "iris_r", passed = reticulate::r_to_py(all(!is.na(iris$Sepal.Width))), check_name = "no_missing_sepal_width_check_r", ) # Ensure that Sepal.Length field does not contain any NAs context$report_asset_check( asset_key = "iris_r", passed = reticulate::r_to_py(all(!is.na(iris$Petal.Length))), check_name = "no_missing_petal_length_check_r", ) # Ensure that Sepal.Width field does not contain any NAs context$report_asset_check( asset_key = "iris_r", passed = reticulate::r_to_py(all(!is.na(iris$Petal.Width))), check_name = "no_missing_petal_width_check_r", ) # Ensure that Species field contains expected set of values context$report_asset_check( asset_key = "iris_r", passed = reticulate::r_to_py(all(iris$Species %in% c("setosa", "versicolor", "virginica"))), check_name = "species_name_check_r", ) # Write iris data to csv so it can be read by a downstream Python asset readr::write_csv(iris, glue::glue("{output_dir}/iris.csv")) })

Based on the provided R script, it seems that you are using the

reticulate

package to interface with Python and the

dagster_pipes

library to report asset materializations and checks back to Dagster from an R script. The script uses the

with

context provided by

reticulate

to ensure that the

open_dagster_pipes()

context is properly managed. From the error message you received, it appears that Dagster is expecting certain outputs from the

iris_r

operation, but they are not being yielded or returned as expected. The error message specifically mentions the need to use

yield from pipes_session.get_results()

or

return <PipesClient>.run(...).get_results()

. However, in the R script you provided, there is no explicit yielding or returning of results that would correspond to these Python-specific constructs. Here are some steps to address the issue: 1. Ensure Proper Asset Reporting: Verify that the

context$report_asset_materialization()

and

context$report_asset_check()

calls are correctly reporting the materializations and checks back to Dagster. These calls should correspond to the expected outputs mentioned in the error message. 2. Check Environment Variables: The script is using environment variables like

MY_ENV_VAR_IN_SUBPROCESS

and

OUTPUT_DIR

. Ensure that these are correctly set in your Ubuntu environment, as they might differ from your Windows setup. 3. Verify Output Directory: The script writes the

iris

dataset to a CSV file in the

output_dir

. Make sure that this directory exists and is writable on your Ubuntu machine. 4. Review Reticulate Configuration: Since you're using

reticulate

to bridge R and Python, ensure that the Python environment configured for

reticulate

has the correct version of

dagster_pipes

installed and is the same environment where your Dagster instance is running. 5. Check for OS-Specific Issues: There might be differences in how file paths or environment variables are handled between Windows and Ubuntu. Review the script for any such discrepancies. 6. Debugging Logs: Add additional logging statements in your R script to capture the state of the script at various points. This can help you understand if all parts of the script are executing as expected. 7. Consult Dagster Documentation: Review the Dagster documentation for any additional requirements or best practices when using Dagster Pipes with R scripts. If after checking these steps the issue persists, you may need to look into the specifics of how Dagster Pipes integrates with R and whether there are additional steps required to ensure that the expected outputs are correctly communicated back to Dagster. It might also be helpful to consult the Dagster community or support channels for guidance on using Dagste

@Scout, i'm not really sure what is needed when it comes to returning/yielding the context$report_asset_materialization() function... Do I need to combine this with R's return() function?

In the context of Dagster and its integration with R through the

reticulate

package, the

context$report_asset_materialization()

function is used to report the materialization of an asset to Dagster. This function is part of the Dagster Pipes API, which allows you to communicate with Dagster from an external process, such as an R script. In Python, when working with Dagster's asset materializations within an op or a graph, you might yield

AssetMaterialization

events or return a

MaterializeResult

object to indicate that an asset has been materialized. However, in R, the concept of yielding is not directly applicable as it is in Python. When using the

reticulate

package to interface with Python, you typically call Python functions and methods directly from R without the need to use R's

return()

function in the same way you would in a pure R script. The

context$report_asset_materialization()

call is made within the

with

block provided by

reticulate

, and it should handle the communication with Dagster internally. Here's what you should consider: - Direct Calls: The

context$report_asset_materialization()

function is called directly within the

with

block, and you do not need to use R's

return()

function to return its result. The

reticulate

package should handle the communication with the Python context created by Dagster Pipes. - Implicit Reporting: The

context$report_asset_materialization()

function implicitly reports the materialization event to Dagster when it is called. You do not need to explicitly yield or return its result for Dagster to recognize the materialization. - Error Handling: If you encounter errors related to asset materialization not being reported, ensure that the

context$report_asset_materialization()

function is being called correctly and that any required parameters are being passed. - Documentation and Examples: Consult the Dagster documentation for examples and guidance on using Dagster Pipes with languages other than Python, such as R. The documentation may provide specific instructions or best practices for reporting asset materializations from R scripts. If you continue to face issues with reporting asset materializations from your R script, it may be helpful to reach out to the Dagster community for support, as they may have more experience with similar integrations and can provide more targeted assistance.

@Scout, here is what the asset definition looks like on the python/dagster side. Are there any issues with this and how it's interacting with my R script? I'm still not understanding why this code works on my windows machine, but fails on my ubuntu machine... @asset( config_schema={"output_dir": Field(String, default_value="./data")}, check_specs=[ AssetCheckSpec(name="no_missing_sepal_length_check_r", asset="iris_r"), AssetCheckSpec(name="no_missing_sepal_width_check_r", asset="iris_r"), AssetCheckSpec(name="no_missing_petal_length_check_r", asset="iris_r"), AssetCheckSpec(name="no_missing_petal_width_check_r", asset="iris_r"), AssetCheckSpec(name="species_name_check_r", asset="iris_r"), ], ) def iris_r( context: AssetExecutionContext, pipes_subprocess_client: PipesSubprocessClient, ) -> MaterializeResult: output_dir = context.op_config["output_dir"] cmd = [shutil.which("Rscript"), file_relative_path(file, "./R/iris.R")] return pipes_subprocess_client.run( command=cmd, context=context, env={ "MY_ENV_VAR_IN_SUBPROCESS": "This is an environment variable passed from Dagster to R!", "OUTPUT_DIR": output_dir, }, ).get_materialize_result()

The Python code you've provided for the

iris_r

asset definition in Dagster looks correct in terms of its structure and usage of the

PipesSubprocessClient

to run an R script. The asset is configured with a schema for an

output_dir

and a series of

AssetCheckSpec

objects that correspond to checks you expect to be reported from the R script. Here are a few things to consider that might be causing the discrepancy in behavior between your Windows and Ubuntu machines: 1. Environment Variables: The environment variables

MY_ENV_VAR_IN_SUBPROCESS

and

OUTPUT_DIR

are being set in the

env

parameter of the

run

method. Ensure that these are being correctly used within your R script and that the R environment on your Ubuntu machine is correctly picking up these variables. 2. Rscript Path: The

shutil.which("Rscript")

function is used to find the path to the

Rscript

executable. It's possible that the path to

Rscript

might differ between your Windows and Ubuntu environments, or that

Rscript

might not be in the system's PATH on your Ubuntu machine. Verify that

Rscript

is installed and accessible from the command line on your Ubuntu machine. 3. File Paths: The

file_relative_path

function is used to construct the path to the

iris.R

script relative to the Python file. Ensure that the file structure is the same on both machines and that the relative path is correct. 4. Filesystem Permissions: Differences in filesystem permissions between Windows and Ubuntu could affect the ability to read or write files. Ensure that the

output_dir

is writable on your Ubuntu machine. 5. R Package Versions: If your R script relies on specific versions of R packages, ensure that the same versions are installed on both your Windows and Ubuntu machines. 6. R Configuration: The R environment itself might be configured differently on your Ubuntu machine. Check for any

.Rprofile

or

.Renviron

files that might set different options or environment variables in R. 7. Error Messages: If there are any error messages or logs produced by the R script when it runs on your Ubuntu machine, they could provide clues as to what might be going wrong. 8. Debugging: Add additional logging or debugging statements in both the R script and the Python asset definition to trace the execution flow and identify where it might be diverging between the two environments. Given that the code works on your Windows machine, the issue is likely related to the environment setup on your Ubuntu machine rather than the code itself. By checking the above points, you should be able to identify and resolve the discrepancy. If the issue persists, consider reaching out to the Dagster community for further assistance, as they may have insights specific to runnin

@Scout, would the location of context$report_asset_materialization() matter within my with block? Does it make more sense to move this to the end of the with block (after asset checks)? Does it even matter?

In the context of using

dagster_pipes

with R, the

context$report_asset_materialization()

function is used to report the materialization of an asset to Dagster. The location of this function call within the

with

block can be important, depending on what you intend to communicate to Dagster and when. According to the documentation, when you report an asset materialization, you are indicating to Dagster that a certain asset has been created or updated. This is typically done after the actual work to create or update the asset has been completed. Therefore, it would make sense to call

context$report_asset_materialization()

after the asset has been fully processed and any checks have been performed. Here's why the order can matter: - Asset Checks First: Performing and reporting asset checks before reporting the materialization allows you to ensure that all checks pass before the asset is considered materialized. If any checks fail, you might decide to handle these failures accordingly before reporting the materialization. - Materialization as a Final Step: Reporting the materialization after all checks have been reported provides a clear indication that the asset is in a good state and that all checks have passed. It acts as a final step in the asset creation or update process. - Logical Flow: From a logical perspective, it makes sense to report the materialization after the asset's data has been written to its destination (e.g., a file has been saved, a database has been updated) and after all relevant checks have been performed. In summary, while it might not be strictly required to call

context$report_asset_materialization()

at the end of the

with

block, doing so is a good practice as it reflects the actual sequence of operations: processing the data, performing checks, and then reporting the materialization. This sequence ensures that the asset is in the expected state when it is reported to Dagster. If you are experiencing issues with the order of operations, consider moving the

context$report_asset_materialization()

call to the end of the

with

block, after all checks have been reported, to see if that resolves the issue.