Runner 编写指南

本指南介绍如何实现新的 runner。它面向那些拥有数据处理系统并希望使用它来执行 Beam 管道的人。指南从基础开始，帮助你评估未来的工作。然后，各部分将越来越详细，成为你整个 runner 开发过程中的资源。

涵盖的主题

实现 Beam 原语

除了编码和持久化数据（你的引擎可能已经以某种方式实现了这些功能）之外，你需要做的就是实现 Beam 原语。本部分详细介绍了每个原语，涵盖了可能不明显的内容以及提供的支持代码。

这些原语的设计是为了管道作者，而不是 runner 作者。每个原语都代表了一种不同的概念操作模式（外部 I/O、元素级、分组、窗口化、联合），而不是具体的实现决策。同一个原语可能需要非常不同的实现，具体取决于用户如何实例化它。例如，使用状态或计时器的 ParDo 可能需要键分区，具有推测性触发的 GroupByKey 可能需要更昂贵或更复杂的实现。

如果你没有实现其中的一些功能怎么办？

没关系！你不必一次性完成所有操作，甚至可能有些功能对你的 runner 来说永远没有意义。我们在 Beam 网站上维护了一个 功能矩阵，以便你可以告诉用户你支持什么功能。当你收到一个 Pipeline 时，你应该遍历它并确定你是否可以执行你遇到的每个 DoFn。如果你无法执行管道中的某个 DoFn（或者你的 runner 缺少其他任何要求），你应该拒绝该管道。在你的原生环境中，这可能看起来像抛出 UnsupportedOperationException。Runner API RPC 将明确说明这一点，以实现跨语言的可移植性。

实现 Impulse 原语

Impulse 是一个 PTransform，它不接受任何输入，并在管道生命周期内产生一个输出，该输出应该是全局窗口中带有最小时间戳的空字节。在使用标准窗口化值编码器编码时，它的编码值为 7f df 3b 64 5a 1c ac 09 00 00 00 01 0f 00。

尽管 Impulse 通常不被用户调用，但它是唯一的基本原语操作，其他基本操作（如 Read 和 Create）都是通过 Impulse 后跟一系列（可能是可拆分的）ParDo 构建的复合操作。

实现 ParDo 原语

ParDo 原语描述了 PCollection 的元素级转换。ParDo 是最复杂的原语，因为它描述了任何元素级处理。除了像函数式编程中的标准 map 或 flatMap 这样的非常简单的操作外，ParDo 还支持多个输出、侧输入、初始化、刷新、拆卸和有状态处理。

应用于每个元素的 UDF 称为 DoFn。DoFn 的确切 API 可能因语言/SDK 而异，但通常遵循相同的模式，因此我们可以用伪代码来讨论它。我还会经常引用 Java 支持代码，因为我熟悉它，并且我们当前和未来的大多数 runner 都是基于 Java 的。

通常，与逐个对整个输入数据集应用一系列 ParDo 相比，将几个 ParDo 合并到单个可执行阶段中通常更有效，该阶段包含一系列（通常是 DAG）映射操作。除了 ParDo 之外，窗口操作、本地（预先或事后 GBK）组合操作以及其他映射操作也可以合并到这些阶段中。

由于 DoFns 可以在与 runner 本身不同的语言或需要不同环境中执行代码，因此 Beam 提供了以跨进程方式调用这些代码的能力。这是 Beam Fn API 的核心，后面会详细介绍。但是，当环境兼容时，runner 可以将此用户代码在进程中调用（为了简单或效率）是完全可以接受的。

捆绑

为了正确性，DoFn 应该 代表一个元素级函数，但在大多数 SDK 中，它是一个长期存在的对象，它以称为捆绑的小组形式处理元素。

你的 runner 决定每个捆绑包含多少元素以及哪些元素，甚至可以在处理过程中动态决定当前捆绑“结束”。捆绑的处理方式与 DoFn 的整个生命周期相关联。

通常，尽可能使用最大的捆绑可以提高吞吐量，这样初始化和最终化成本就可以在许多元素上摊销。但是，如果你的数据作为流到达，那么你将需要终止一个捆绑以获得适当的延迟，因此捆绑可能只有几个元素。

捆绑是 Beam 中的承诺单位。如果在处理捆绑时遇到错误，则 runner 必须丢弃该捆绑的所有先前输出（包括对状态或计时器的任何修改），并重试整个捆绑。捆绑成功完成时，其输出、以及任何状态/计时器修改和水印更新，必须以原子方式提交。

DoFn 生命周期

许多 SDK 中的 DoFns 除了标准的元素级 process 调用之外，还有几个方法，例如 setup、start_bundle、finish_bundle、teardown 等。通常，当你从标准捆绑处理器（通过 FnApi 或直接使用 BundleProcessor (java (python)) 调用一个或多个 DoFn 时，应该为你处理适当的 生命周期 调用。与 SDK 无关的 runner 不应该直接处理这些细节。

侧输入

主要设计文档：https://s.apache.org/beam-side-inputs-1-pager

侧输入是 PCollection 窗口的全局视图。这将它与逐个元素处理的主输入区分开来。SDK/用户适当地准备一个 PCollection，runner 对其进行物化，然后 runner 将其馈送到 DoFn。

与由 runner 推送到 ParDo（通常通过 FnApi 数据通道）的主输入数据不同，侧输入数据是由 ParDo 从 runner 拉取的（通常通过 FnAPI 状态通道）。

侧输入通过特定的 access_pattern 访问。目前，StandardSideInputTypes 协议中列出了两种访问模式：beam:side_input:iterable:v1 表示 runner 必须返回与特定窗口相对应的 PCollection 中的所有值，beam:side_input:multimap:v1 表示 runner 必须返回与特定键和窗口相对应的所有值。能够有效地提供这些访问模式可能会影响 runner 对此 PCollection 的物化方式。

可以通过查看 ParDo 转换的 ParDoPayload 中的 side_inputs 映射来检测侧输入。ParDo 操作本身负责调用 window_mapping_fn（在调用 runner 之前）和 view_fn（在 runner 返回的值上），因此 runner 不需要关心这些字段。

当需要侧输入但侧输入在给定窗口中没有与之关联的数据时，该窗口中的元素必须被延迟，直到侧输入有一些数据，或者水印已充分前进，以至于我们可以确定该窗口中将没有数据。PushBackSideInputDoFnRunner 是实现这一点的一个例子。

状态和计时器

主要设计文档：https://s.apache.org/beam-state

当 ParDo 包含状态和计时器时，它在你的 runner 上的执行通常会有很大不同。特别是，当捆绑完成时必须持久化状态，并在未来的捆绑中检索它。设置的计时器还必须在水印足够前进时注入到未来的捆绑中。

状态和计时器按键和窗口进行分区，也就是说，处理给定键的 DoFn 必须对所有共享此键的元素具有状态和计时器的一致视图。你可能需要或希望显式地对数据进行混洗以支持这一点。一旦水印超过了窗口的结束（加上允许的延迟，如果有的话），与该窗口关联的状态就可以被删除。

状态设置和检索是在 FnAPI 状态通道上执行的，而计时器设置和触发是在 FnAPI 数据通道上执行的。

可拆分 DoFn

主设计文档: https://s.apache.org/splittable-do-fn

可分割的 DoFn 是 ParDo 的泛化，适用于可以并行执行的高扇出映射。此类操作的典型示例是从文件读取，其中单个文件名（作为输入元素）可以映射到该文件中包含的所有元素。DoFn 被认为是可分割的，因为表示单个文件的元素可以被分割（例如，分成该文件的范围）以供不同的工作者处理（例如，读取）。这种原语的强大功能在于，这些分割可以动态发生，而不仅仅是静态发生（即提前发生），从而避免了过度分割或分割不足的问题。

本文档不包含对可分割 DoFn 的完整解释，但这里简要概述了它与执行相关的部分。

可分割的 DoFn 可以通过在元素内部和元素之间进行分割来参与动态分割协议。动态分割由运行器在控制通道上发出 ProcessBundleSplitRequest 消息触发。SDK 将承诺仅处理指示的元素的一部分，并将剩余部分（即未处理部分）的描述返回给运行器，以在 ProcessBundleSplitResponse 中进行调度（例如，在不同的工作者上或作为不同捆绑的一部分）。

可分割的 DoFn 也可以启动自己的分割，表明它已尽可能地处理了一个元素（例如，当尾随文件时），但还有更多元素需要处理。这些情况最常发生在读取无界源时。在这种情况下，表示延迟工作的一组元素将通过 ProcessBundleResponse 的 residual_roots 字段传回。在将来，运行器必须使用 residual_roots 中给出的元素重新调用相同的操作。

实现 GroupByKey（和窗口）原语

GroupByKey 操作（有时简称 GBK）按键和窗口对键值对的 PCollection 进行分组，并根据 PCollection 的触发配置发出结果。

它比简单地将具有相同键的元素放在一起要复杂得多，并且使用了 PCollection 的窗口策略中的许多字段。

按编码字节分组

对于键和窗口，您的运行器将它们视为“仅仅是字节”。因此，您需要以与按这些字节分组一致的方式进行分组，即使您对所涉及的类型有一些特殊知识。

您正在处理的元素将是键值对，您需要提取键。因此，键值对的格式是 标准化的，并在所有 SDK 中共享。有关二进制格式的文档，请参阅 Java 中的 KvCoder 或 Python 中的 TupleCoder。

窗口合并

除了按键分组外，您的运行器还必须按元素的窗口对其进行分组。WindowFn可以选择声明它在每个键的基础上合并窗口。例如，如果相同键的会话窗口重叠，它们将被合并。因此，您的运行器必须在分组期间调用 WindowFn 的合并方法。

通过 GroupByKeyOnly + GroupAlsoByWindow 实现

Java 和 Python 代码库包含对实现完整 GroupByKey 操作的一种特别常见方式的支持代码：首先对键进行分组，然后按窗口进行分组。对于合并窗口，这实际上是必需的，因为合并是按键进行的。

通常，按时间戳顺序呈现值集可以更有效地将这些值分组到其最终窗口中。

丢弃延迟数据

主设计文档: https://s.apache.org/beam-lateness

如果输入 PCollection 的水印至少超过窗口结束时间，输入 PCollection 允许的延迟时间，则 PCollection 中的窗口就会过期。

过期窗口的数据可以在任何时间丢弃，并且应该在 GroupByKey 中丢弃。如果您使用的是 GroupAlsoByWindow，那么在执行此转换之前即可丢弃数据。如果您在 GroupByKeyOnly 之前丢弃数据，可能会减少数据混洗量，但只有对于非合并窗口才安全，因为看似过期的窗口可能会合并而变得不再过期。

触发

主设计文档: https://s.apache.org/beam-triggers

输入 PCollection 的触发器和累积模式指定何时以及如何从 GroupByKey 操作发出输出。

在 Java 中，GroupAlsoByWindow 实现、ReduceFnRunner（旧名称）和 TriggerStateMachine 中有大量支持代码用于执行触发器，这是一种显而易见的方式，可以将所有触发器实现为基于元素和计时器的事件驱动机器。在 Python 中，这是通过 TriggerDriver 类支持的。

TimestampCombiner

当从多个输入生成一个聚合输出时，GroupByKey 操作必须为组合选择一个时间戳。为此，首先 WindowFn 有机会移位时间戳——这是为了确保水印不会阻止窗口（如滑动窗口）的进度（详细信息不在本文档范围内）。然后，需要合并移位后的时间戳——这由 TimestampCombiner 指定，它可以选择其输入的最小值或最大值，或者忽略输入并选择窗口结束时间。

实现 Window 原语

窗口原语将 WindowFn UDF 应用于每个输入元素，以将其放置到其输出 PCollection 的一个或多个窗口中。请注意，此原语通常还会配置 PCollection 的窗口策略的其他方面，但您的运行器接收到的完全构建的图将已经为每个 PCollection 具有完整的窗口策略。

要实现此原语，您需要对每个元素调用提供的 WindowFn，它将返回该元素要作为输出 PCollection 部分的某些窗口集。

大多数运行器通过将这些改变窗口的映射与 DoFns 融合来实现这一点。

实现注意事项

“窗口”只是一个具有“最大时间戳”的第二个分组键。它可以是任何任意的用户定义类型。WindowFn 提供了窗口类型的编码器。

Beam 的支持代码提供 WindowedValue，它是在多个窗口中元素的压缩表示形式。您可能想要使用它，或者使用您自己的压缩表示形式。请记住，它只是同时表示多个元素；不存在元素“在多个窗口中”这样的概念。

对于全局窗口中的值，您可能想要使用更进一步的压缩表示形式，它根本不包含窗口。

我们提供了具有这些优化的编码器，例如 PARAM_WINDOWED_VALUE，它可以用于减少序列化数据的规模。

将来，此原语可能会被弃用，因为它可以在 ParDo 的功能增强以允许输出到新窗口的情况下，作为 ParDo 来实现。

实现 Flatten 原语

这个很简单——将一组有限的 PCollections 作为输入，并输出它们的集合并，保持窗口完整。

为了使此操作有意义，SDK 负责确保窗口策略兼容。

还要注意，所有 PCollections 的编码器都没有必要相同。如果您的运行器想要要求它们相同（以避免繁琐的重新编码），您必须自己强制执行。或者，您也可以只实现快速路径作为优化。

特别说明：Combine 复合

CombinePerKey 是一种复合转换，几乎总是由运行器特殊对待，它将一个关联的、可交换的操作符应用于 PCollection 的元素。此复合不是原语。它使用 ParDo 和 GroupByKey 实现，因此您的运行器无需处理它——但它确实包含您可能想用于优化的其他信息：关联的、可交换的操作符，称为 CombineFn。

通常，运行器会使用组合器提升来实现这一点，其中一个新的操作将在 GroupByKey 之前放置，执行部分（捆绑内）组合，这通常还需要稍微修改 GroupByKey 之后的内容。此转换的示例可以在 Python 或 go 中找到此优化的实现。生成的 GroupByKey 前后操作通常会与 ParDo 融合并按上述方式执行。

使用管道

当您从用户那里收到一个管道时，您需要对其进行翻译。有关 Beam 管道如何表示的解释，可以 在这里找到，它补充了 官方的协议声明。

测试你的 runner

Beam Java SDK 和 Python SDK 具有运行器验证测试套件。配置可能比本文档发展得更快，因此请检查其他 Beam 运行器的配置。但是请注意，我们有测试，您可以非常容易地使用它们！要在使用 Gradle 的基于 Java 的运行器中启用这些测试，请扫描 SDK 的依赖项以查找具有 JUnit 类别 ValidatesRunner 的测试。

task validatesRunner(type: Test) {
  group = "Verification"
  description = "Validates the runner"
  def pipelineOptions = JsonOutput.toJson(["--runner=MyRunner", ... misc test options ...])
  systemProperty "beamTestPipelineOptions", pipelineOptions
  classpath = configurations.validatesRunner
  testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs)
  useJUnit {
    includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
  }
}

在其他语言中启用这些测试尚未得到探索。

将你的 runner 与 SDK 良好集成

无论您的运行器是否基于与 SDK 相同的语言（如 Java），如果您想让该 SDK（如 Python）的用户使用它，您都需要提供一个垫片来从其他 SDK 调用它。

与 Java SDK 集成

允许用户向你的 runner 传递选项

配置机制是 PipelineOptions，它是一个接口，其工作方式与普通的 Java 对象完全不同。忘记您所知道的一切，遵循规则，PipelineOptions 会对您很好。

您必须为您的运行器实现一个子接口，并使用匹配的名称提供 getter 和 setter，如下所示

public interface MyRunnerOptions extends PipelineOptions {
  @Description("The Foo to use with MyRunner")
  @Required
  public Foo getMyRequiredFoo();
  public void setMyRequiredFoo(Foo newValue);

  @Description("Enable Baz; on by default")
  @Default.Boolean(true)
  public Boolean isBazEnabled();
  public void setBazEnabled(Boolean newValue);
}

您可以设置默认值等。有关详细信息，请参阅 javadoc。当您的运行器使用 PipelineOptions 对象实例化时，您可以通过 options.as(MyRunnerOptions.class) 访问您的接口。

要使这些选项在命令行上可用，您需要使用 PipelineOptionsRegistrar 注册您的选项。如果您使用 @AutoService，则这很简单。

@AutoService(PipelineOptionsRegistrar.class)
public static class MyOptionsRegistrar implements PipelineOptionsRegistrar {
  @Override
  public Iterable<Class<? extends PipelineOptions>> getPipelineOptions() {
    return ImmutableList.<Class<? extends PipelineOptions>>of(MyRunnerOptions.class);
  }
}

将你的 runner 注册到 SDK 以供命令行使用

要使您的运行器在命令行上可用，您需要使用 PipelineRunnerRegistrar 注册您的选项。如果您使用 @AutoService，则这很简单。

@AutoService(PipelineRunnerRegistrar.class)
public static class MyRunnerRegistrar implements PipelineRunnerRegistrar {
  @Override
  public Iterable<Class<? extends PipelineRunner>> getPipelineRunners() {
    return ImmutableList.<Class<? extends PipelineRunner>>of(MyRunner.class);
  }
}

与 Python SDK 集成

在 Python SDK 中，代码的注册不是自动的。因此，在创建新运行器时，需要记住一些事项。

对新运行器包的任何依赖项都应该是选项，因此请在 setup.py 中创建一个新的目标，即新运行器需要的 extra_requires。

所有运行器代码都应该放在 apache_beam/runners 目录中的自己的包中。

在 runner.py 的 create_runner 函数中注册新的运行器，以便部分名称与要使用的正确类匹配。

Python 运行器也可以通过其完全限定的名称来识别（例如，在传递运行器参数时），无论它们是否位于 Beam 存储库中。

编写与 SDK 无关的 runner

使您的运行器与 SDK 独立，能够运行用其他语言编写的管道有两个方面：Fn API 和 Runner API。

Fn API

设计文档

为了运行用户的管道，您需要能够调用他们的 UDF。Fn API 是 Beam 标准 UDF 的 RPC 接口，使用 gRPC 上的协议缓冲区实现。

Fn API 包括

您完全可以也使用您语言的 SDK 进行实用程序代码，或为相同语言的 UDF 提供优化的捆绑包处理实现。

Runner API

该运行器 API是管道的一种与 SDK 无关的模式，以及用于启动管道和检查作业状态的 RPC 接口。通过仅通过 Runner API 接口检查管道，您可以消除运行器对管道分析和作业转换语言的 SDK 的依赖关系。

要执行此类与 SDK 无关的管道，您需要支持 Fn API。UDF 作为函数规范（通常只是特定语言的序列化字节）和可以执行它的环境规范（本质上是特定 SDK）嵌入到管道中。到目前为止，此规范预计是托管 SDK Fn API 框架的 Docker 容器的 URI。

您完全可以也使用您语言的 SDK，它可能提供有用的实用程序代码。

管道的与语言无关的定义通过协议缓冲区模式进行描述，以下列出供参考。但是，您的运行器不需要直接操作协议缓冲区消息。相反，Beam 代码库提供了用于处理管道的实用程序，因此您不需要了解管道是否曾经被序列化或传输，或者最初可能使用什么语言编写。

Java

如果您的运行器是基于 Java 的，那么以与 SDK 无关的方式与管道交互的工具位于beam-runners-core-construction-java工件中，位于org.apache.beam.sdk.util.construction命名空间中。这些实用程序的命名是一致的，如下所示

通过仅通过这些类检查转换，您的运行器将不会依赖于 Java SDK 的细节。

Runner API 协议

该运行器 API指的是 Beam 模型中概念的一种特定表现形式，作为协议缓冲区模式。即使您不应该直接操作这些消息，了解构成管道的规范数据也很有帮助。

大多数 API 与高级描述完全相同；您无需了解所有低级细节即可开始实现运行器。

对您来说，Runner API 最重要的收获是它是 Beam 管道的与语言无关的定义。您可能总是通过特定 SDK 的支持代码与这些定义进行交互，这些代码使用合理的习惯用法 API 包装这些定义，但始终意识到这是规范，任何其他数据并非必然是管道固有的，而是可能特定于 SDK 的增强功能（或错误！）。

管道中的 UDF 可以针对任何 Beam SDK 编写，甚至可以在同一个管道中针对多个 SDK 编写。因此，我们将从这里开始，采用自下而上的方法来了解 UDF 的协议缓冲区定义，然后再回到更高级别、大部分明显的记录定义。

FunctionSpec 协议

跨语言可移植性的核心是FunctionSpec。这是函数的与语言无关的规范，按照通常的编程意义，包括副作用等。

message FunctionSpec {
  string urn;
  bytes payload;
}

一个FunctionSpec包含一个标识函数的 URN 以及一个任意固定参数。例如，（假设的）“max” CombineFn 可能具有 URNbeam:combinefn:max:0.1和一个指示按什么比较取最大值的参数。

对于使用特定语言的 SDK 构建的管道中的大多数 UDF，URN 将表明 SDK 必须对其进行解释，例如beam:dofn:javasdk:0.1或beam:dofn:pythonsdk:0.1。该参数将包含序列化代码，例如 Java 序列化的DoFn或 Python 腌制的DoFn。

一个FunctionSpec不仅仅用于 UDF。它只是命名/指定任何函数的通用方法。它也用作PTransform的规范。但是当在PTransform中使用时，它描述了从PCollection到PCollection的函数，并且不能特定于 SDK，因为运行器负责评估转换并生成PCollections。

不用说，并非所有环境都能反序列化所有函数规范。为此原因，PTransform具有一个environment_id参数，该参数至少指示一个能够解释包含的 URN 的环境。这是对 Pipeline proto 中环境映射中的环境的引用，通常由 docker 映像（可能带有一些额外的依赖项）定义。可能还有其他环境也能够做到这一点，运行器可以自由使用它们，如果它有这方面的知识。

基本转换有效载荷协议

原始转换的有效负载只是其规范的协议序列化。我将重点介绍重要的部分以显示它们是如何组合在一起的，而不是在此处重现它们的完整代码。

值得再次强调的是，虽然您可能不会直接与这些有效负载进行交互，但它们是转换中固有的唯一数据。

ParDoPayload 协议

一个ParDo转换将其DoFn放在一个SdkFunctionSpec中，然后为其其他功能提供与语言无关的规范 - 侧输入、状态声明、计时器声明等。

message ParDoPayload {
  FunctionSpec do_fn;
  map<string, SideInput> side_inputs;
  map<string, StateSpec> state_specs;
  map<string, TimerSpec> timer_specs;
  ...
}

CombinePayload 协议

Combine不是原始的。但是非原始类型完全可以携带其他信息以实现更好的优化。Combine转换携带的最重要的事情是CombineFn在一个SdkFunctionSpec记录中。为了有效地执行所需的优化，还需要知道中间累积的编码器，因此它也包含对该编码器的引用。

message CombinePayload {
  FunctionSpec combine_fn;
  string accumulator_coder_id;
  ...
}

PTransform 协议

一个PTransform是从PCollection到PCollection的函数。这在协议中使用 FunctionSpec 表示。

message PTransform {
  FunctionSpec spec;
  repeated string subtransforms;

  // Maps from local string names to PCollection ids
  map<string, bytes> inputs;
  map<string, bytes> outputs;
  ...
}

如果一个PTransform是复合的，它可能具有子转换，在这种情况下，FunctionSpec可以省略，因为子转换定义了它的行为。

输入和输出PCollections是无序的，并且通过本地名称引用。SDK 决定这个名称是什么，因为它可能会嵌入到序列化的 UDF 中。

一个理解给定PTransform规范的运行器（无论是原始的还是复合的），如其FunctionSpec所定义，可以自由地将其替换为具有相同语义的另一个PTransform（或集合）。这通常是处理CombinePerKey的方式，但也可以进行许多其他替换。

PCollection 协议

一个PCollection只存储一个编码器、窗口策略和它是否是有界的。

message PCollection {
  string coder_id;
  IsBounded is_bounded;
  string windowing_strategy_id;
  ...
}

Coder 协议

这是一个非常有趣的协议。编码器是一个参数化的函数，它可能只被特定 SDK 理解，因此是一个FunctionSpec，但也可能具有完全定义它的组件编码器。例如，ListCoder只是一个元格式，而ListCoder(VarIntCoder)是一个完全指定的格式。

message Coder {
  FunctionSpec spec;
  repeated string component_coder_ids;
}

有大量标准编码器为大多数（如果不是全部）SDK 理解。使用这些可以实现跨语言转换。

Jobs API RPC

概述 规范

虽然您语言的 SDK 可能可以防止您直接接触 Runner API 协议，但您可能需要为您的运行器实现适配器，以将其公开给另一种语言。这允许 Python SDK 调用 Java 运行器，反之亦然。这种典型的实现可以在local_job_service.py中找到，该实现直接用于直接为几个用 Python 实现的运行器提供服务。

RPC 本身必然会遵循 PipelineRunner 和 PipelineResult 的现有 API，但经过修改，使其成为最小的后端通道，而不是丰富且方便的 API。

这方面的一个关键部分是工件 API，它允许运行器获取和部署二进制工件（如 jar、pypi 包等），这些工件列在各种环境的依赖项中，并且可能具有各种表示形式。这是在提交管道后但执行管道之前调用的。提交管道的 SDK 充当接收请求的运行器的工件服务器，而运行器又充当托管用户 UDF 的工作者（环境）的工件服务器。