How can I pass dynamically generated inputs into a k8s_job_op

[dagsir[bot]]

I have a binary command which I am going to pack into a container and run as an op (possibly using k8s_job_op). A challenge I'm facing is that my inputs for the binary are dynamically generated outputs from other ops (they're files). Is there a way to pass these files as inputs to the k8s_job_op, or would I need to write them to a volume first?

---

The question was originally asked in Dagster Slack.

[prha]

In order to support this, rather than using k8s_job_op, you could use execute_k8s_job (docs link) and use your own op definition.

Example:

@op
def my_custom_op(context):
    execute_k8s_job(...) # does something
    return "my/file/path"

@op
def my_second_custom_op(context, path: str):
    execute_k8s_job(...) # does something else, maybe using path

@job
def my_job_with_custom_ops():
    my_second_custom_op(my_custom_op())