Executing R scripts from Dagster

[philiporlando]

I'm currently exploring Dagster for data orchestration and find myself in need of some community wisdom. My primary goal is to execute R scripts within the Dagster environment, a task I previously accomplished using Airflow's BashOperator and DockerOperator.

In Airflow, these operators provided a straightforward way to run R scripts either directly through shell commands or within Docker containers. I'm looking for Dagster's equivalents to these operators. Specifically, I'm interested in:

1. How to execute R scripts in Dagster, comparable to using Airflow's BashOperator. I'd like to understand the best practices for running shell commands or scripts, especially for R scripts.

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    # other args...
}

dag = DAG('r_script_bash_operator', default_args=default_args, schedule_interval='@daily')

run_r_script = BashOperator(
    task_id='run_r_script',
    bash_command='Rscript /path/to/your_script.R ',
    dag=dag,
)

run_r_script

2. The Dagster equivalent of Airflow's DockerOperator. My use case often requires running scripts in isolated environments, and I'm curious about executing R scripts within Docker containers managed by Dagster.

from airflow import DAG
from airflow.operators.docker_operator import DockerOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    # other args...
}

dag = DAG('r_script_docker_operator', default_args=default_args, schedule_interval='@daily')

run_r_script_in_docker = DockerOperator(
    task_id='run_r_script_in_docker',
    image='r-base',  # Make sure this image has R installed
    command='Rscript /path/to/your_script.R',
    docker_url='unix://var/run/docker.sock',  # Update as needed
    network_mode='bridge',
    dag=dag,
)

run_r_script_in_docker

I would greatly appreciate any examples, tips, or pointers to relevant documentation. Insights on capturing logging, metadata, and results while executing these scripts would also be incredibly helpful.

Thank you in advance for your guidance and support.

---

This is related to #2624.

[PedramNavid]

Hi @philiporlando and fellow R-user.

You have a few options, but I think the simplest would be to use the dagster-docker op.

From the docs:

from dagster_docker import docker_container_op

from dagster import job

first_op = docker_container_op.configured(
    {
        "image": "busybox",
        "command": ["echo HELLO"],
    },
    name="first_op",
)
second_op = docker_container_op.configured(
    {
        "image": "busybox",
        "command": ["echo GOODBYE"],
    },
    name="second_op",
)

@job
def full_job():
    second_op(first_op())

For more complex integrations, you might find Dagster Pipes interesting as well: https://docs.dagster.io/guides/dagster-pipes

Let me know if this is helpful or if you have other questions.

[philiporlando]

@PedramNavid, thanks! I'll work on at least standing up a git repo to track this work.

[philiporlando]

Ok, here's an update re passing data between R and Dagster:

Dagster Pipes does not currently support streaming data objects directly between external processes and Dagster assets. - Scout

Since Dagster Pipes does not currently support in-memory data passing between the external process and Dagster, I don't think it will be possible to decouple IO from business logic when defining assets that execute external R code. Not even reticulate::r_to_py() can save us here! Instead, we'd need to serialize the data to an intermediate storage system that both languages can understand. For example, the iris data.frame could be serialized to CSV, JSON, Parquet, Arrow, DuckDB, etc., and then have a downstream Python asset read and deserialize the data from that file.

write.csv(iris, "iris.csv", row.names = FALSE)

import pandas as pd

@asset
def my_python_asset():
    df = pd.read_csv("iris.csv")
    # Perform operations on the dataframe
    return df

Despite this limitation, I still feel like Dagster Pipes is the best way to execute R code from Dagster because you can still pass context, environment variables, logs, and metadata back and forth, which gives me way more visibility about the R asset that is being produced-- and all from within the Dagster UI. I'd call that a huge win!

[alangenfeld]

in-memory data passing

While its not supported in pipes directly, it might be possible to

1. create a SharedMemory on the orchestration side
2. pass the name using pipes extras / env var / etc.
3. open the SharedMemory by name in the R process using the name and write data in to it
4. read the data out on the orchestration side when the external process finishes

Some limitations could cause this path to be infeasible, but could be worth exploring

https://docs.python.org/3/library/multiprocessing.shared_memory.html

[philiporlando]

Thanks for the suggestion, @alangenfeld! I'll give this a shot and see what happens.

[philiporlando]

I created this repo to further delve into the integration of Dagster and R. It includes some examples of using asset checks with R. I'll be sure to include any other progress I make with in-memory data passing there too.