当至少给出一个必需的输入时执行实体

Question

作为输入，我想根据用户输入检索数据，或者如果没有用户输入，则“随机”从数据库中检索数据。管道的所有其他下游任务都是相同的。

因此，我想创建一个从实体 A 和 B 开始的管道，以及基于实体 A OR 实体 B.
的输入执行的下游实体 C。但是，当在实体 A 和 B 上使用条件输出时，不会执行实体 C，因为一个输入不是由上游实体生成的。

有没有一种我错过的简单方法？

感谢您的帮助。

Answer 1

除非跳过所有扇入的输出，否则不会跳过“扇入”依赖项，因此这是实现此目的的一种方法。

@pipeline
def example():
  maybe_a = A()
  maybe_b = B()
  C(items=[maybe_a, maybe_b])

https://docs.dagster.io/examples/fan_in_pipeline#main

Answer 2

根据@Alex 的建议，这是一个可执行管道：

from dagster import pipeline, solid


@solid()
def A(_):
    return


@solid()
def B(_):
    return

@solid()
def C(context, results):
    context.log.info("a returned {}".format(results[0]))
    context.log.info("b returned {}".format(results[1]))
    context.log.info("compute c")


@pipeline
def my_pipeline():
    a = A()
    b = B()
    C([a, b])

当至少给出一个必需的输入时执行实体

Executing a solid when at least one of the required inputs is given

dagster