Lynxus Python 开发者指南

每当 Java 对象必须对 Python API 可用时，它都通过代理对象来提供。这意味着要从 Python 访问 Java 对象，我们只需调用 Python 代理上的适当方法。当调用此方法时，Python 对象会生成一条消息并通过套接字发送。该消息本质上是对"在对象 XYZ 上调用方法 ABC，使用参数 W、X 和 Y"的编码。

例如，如果我们有一个名为 flowFile 的 InputFlowFile 对象，并且想要获取 filename 属性，可以通过调用：

从 Python API 的角度来看，这就是全部所需。在幕后，会向本地套接字写入一条消息，该消息是对"在 ID 为 679212 的对象上调用 getAttribute 方法，字符串参数为 'filename'"的编码。然后 Java 进程接收此命令，在具有给定标识符的对象上调用指定方法，并将该方法调用的结果写回套接字。因此，Python 端接收到 "filename" 属性的值。

这很重要，因为这意味着在 Java 对象上发生的任何方法调用都必须被序列化、通过套接字写入、反序列化，然后才能调用该方法。然后结果必须被序列化、通过套接字写入，并在 Python 端反序列化结果。虽然这是一个相当高效的过程，但它远不如原生调用方法高效。因此，考虑 Java 对象方法调用的开销很重要。

同样重要的是要理解，每当一个对象作为参数提供给 Python 处理器时，该对象只有在 Java 端使其可用期间才能在 Python 端访问。由于 Java 端不能无限期地存储所有对象，因此必须进行一些清理。此清理在方法调用后立即发生。

这意味着如果 FlowFileTransform 的 transform 方法被调用时传入了 ProcessContext 对象，该对象仅在方法调用期间可用。一旦方法返回（无论成功与否），该对象将不再可用。因此，不应将提供给方法调用的对象存储以供以后使用，例如将值赋给 self.processContext。

引用不再可访问的对象将导致类似于以下的错误：

此提示"Target Object ID does not exist for this gateway…​"表示 Python 代码正在尝试访问不再可访问的 Java 对象。

FlowFileTransform API 提供了一种基于 FlowFile 的属性及其文本或二进制内容来路由和转换 FlowFile 的机制。将其与 RecordTransform API 对比，后者提供了一种路由和转换单个记录（例如 JSON、Avro 或 CSV 记录）的机制。

为了实现 FlowFileTransform API，Python 类必须继承自 lynxusapi.FlowFileTransform 类并实现 transform(ProcessContext, InputFlowFile) 方法，该方法返回 FlowFileTransformResult。

此外，处理器类必须以子类形式提供两条信息：它实现的 Java 接口（将始终为 cn.org.acal.lynxus.python.processor.FlowFileTransform）以及有关处理器的任何详细信息，如版本、描述、可能与处理器关联的关键字/标签等。这些将在下面的 ProcessorDetails 和 Java 内部类 部分中更详细地讨论。

因此，一个简单的实现可能如下所示：

transform 方法需要两个参数：context（类型为 lynxusapi.properties.ProcessContext）和 flowfile（类型为 InputFlowFile）。

返回类型是 FlowFileTransformResult，它指示 FlowFile 应传输到哪个关系、FlowFile 的更新内容以及应添加到 FlowFile（或覆盖）的任何属性。relationship 是必需参数。contents 是可选的。如果不需要更新 FlowFile 的内容，则 contents 应不指定或指定为 None。不应返回原始 FlowFile 内容，因为它与传递 None 具有相同的效果，但会更昂贵，因为内容将被写入 FlowFile。同样，除非有任何属性要添加，否则省略 attributes 更有效。

虽然 FlowFileTransform API 提供了一次操作一个 FlowFile 的能力，但 RecordTransform API 为开发者提供了一次操作单个记录的机会。例如，如果 FlowFile 由许多 JSON 记录组成，则可以使用 RecordTransform 处理器来操作每条单独的记录，而无需担心记录是否共置。此 API 的实现必须继承自 RecordTransform 基类，并且还必须实现以下方法：

def transform(self, context, record, schema, attributemap)

返回 RecordTransformResult 对象。

context 对象与 FlowFileTransform 处理器中使用的相同 ProcessContext 实现（参见 context）。record 是一个 Python 字典，表示要操作的记录。无论记录的来源是 JSON、CSV、Avro 还是其他输入格式，此方法都提供了一个 Python 字典。这使得在 Python 中操作数据变得更加简单，并且意味着代码非常可移植，因为它可以操作任何格式的数据。

关联的 schema 对象是 Java 对象 cn.org.acal.lynxus.record.serialization.RecordSchema 的实例。这为数据提供了模式。但是，对模式的调用必须通过套接字传递到 Java 端，因此是昂贵的。

最后，方法签名提供了 attributemap。此 attributemap 有两个方法：

getAttribute(String name)：返回具有给定名称的 FlowFile 属性的值，如果 FlowFile 没有该名称的属性则返回 None。

getAttributes()：返回一个 Python 字典，其键是 FlowFile 属性的名称，值是关联的属性值。

请注意，这两个方法与上面 InputFlowFile 类中的方法完全相同。这允许将 attributemap 提供给 PythonPropertyValue 以评估表达式语言。例如，我们可以通过调用来确定用于某些操作的记录字段名称：

最后，方法必须返回 lynxusapi.recordtransform.RecordTransformResult 的实例。

RecordTransformResult 构造函数接受四个可选的命名参数：

record：记录的转换版本。如果未提供记录，或提供了 None，则输入记录将从输出中删除。

schema：转换后的模式。如果未指定，将推断模式。但是，如果指定了模式，则模式是约束性的，而非数据。因此，如果模式中缺少某个字段，它将从数据中删除。如果模式中有某个字段而数据中没有对应的值，则该字段将被假定具有 None 值。

relationship：要将记录路由到的关系名称。如果未指定，值将路由到 "success" 关系。但是，实现可以选择公开 "success" 和 "failure" 以外的关系并相应地路由记录。例如，实现可能希望将记录记录到 "valid" 或 "invalid"。

partition：默认情况下，给定传入 FlowFile 中的所有记录将写入单个输出 FlowFile（或者更准确地说，是记录的转换版本，假设结果的 record 字段未返回 None 值）。但是，我们可能希望将传入数据分区到单独的输出 FlowFile 中。例如，我们可能有包含 "country" 字段的传入数据，并希望为每个国家生成单独的输出 FlowFile。在这种情况下，我们将为 partition 参数返回一个 Python 字典，类似于 {'country': record['country']}。如果字典中有多个字段，则字典中的所有字段必须具有相同的值，两条记录才能写入同一个输出 FlowFile。

FlowFileSource API 提供了一种创建 FlowFile 并基于其文本或二进制内容路由它的机制。

为了实现 FlowFileSource API，Python 类必须继承自 lynxusapi.FlowFileSource 类并实现 create(ProcessContext) 方法，该方法返回 FlowFileSourceResult。请注意，FlowFileSource 的 create(ProcessContext) 和 FlowFileTransform 的 transform(ProcessContext, InputFlowFile) 方法之间的区别在于前者不需要 InputFlowFile 对象。这是因为基于 FlowFileSource API 的处理器是"源"处理器，不接受传入连接，但能够自己创建 FlowFile。

基于 FlowFileSource 实现处理器与基于 FlowFileTransform 实现非常相似。一个简单的实现如下所示：

如上所述，create 方法只接受一个参数：context（类型为 lynxusapi.properties.ProcessContext）。

返回类型是 FlowFileSourceResult，它指示 FlowFile 应传输到哪个关系、应添加到 FlowFile 的任何属性以及 FlowFile 的内容。relationship 是必需参数。每个基于 FlowFileSource API 的处理器都有一个 success 关系，可以在处理器的 Python 代码中创建额外的关系。attributes 和 contents 都是可选的。如果未提供 attributes，FlowFile 仍将具有通常的 filename、path 和 uuid 属性，但没有额外的属性。如果未提供 contents，将创建一个没有内容的 FlowFile（只有属性）。如果从 create 方法没有有用的信息可返回，可以使用 return None 代替返回空的 FlowFileSourceResult。当 create() 返回 None 时，处理器不产生任何输出。当没有内容可返回时，让出处理器的资源并且不在处理器的 Yield Duration 定义的时间段内调度处理器运行可能是有用的。这可以通过在处理器的 create 方法中返回 None 之前调用 context.yield_resources() 来实现。

任何软件的一个重要方面是配置它的能力。在 Lynxus 中，处理器通过其属性进行配置。为了公开可用的属性，处理器必须为该属性公开一个 PropertyDescriptor。PropertyDescriptor 包含传达如何配置属性所需的所有信息。

PropertyDescriptor 使用 lynxusapi.properties.PropertyDescriptor 类创建。构造函数需要两个必需的定位参数：name 和 description。所有其他参数都是可选的。

通常，处理器会有多个属性描述符。然后通过在处理器中实现以下方法将这些描述符返回给 Lynxus 框架（无论它是 FlowFileTransform、RecordTransform 还是 FlowFileSource）：

此方法返回 PropertyDescriptors 列表。通常的约定是在处理器的构造函数中创建属性描述符，然后在此方法中返回它们，例如：

然而，有时处理器开发者希望允许用户指定自己的属性。例如，我们可能允许用户输入多个键/值对，其中键是要设置的记录字段的名称，值是要设置的值。为此，我们实现以下方法：

该方法返回一个 PropertyDescriptor。例如：

如果未实现此方法并且用户添加了明确支持的属性以外的属性，处理器将变为无效。当然，我们也可以指定可以使用的显式验证器等。

Lynxus 中的每个处理器都必须将其输出数据路由到某个目的地。在 Lynxus 中，这些目的地称为"关系"。每个处理器负责声明其关系。

FlowFileTransform 和 RecordTransform 处理器都已经有一个名为 original 的关系和一个名为 failure 的关系。original 关系不应被实现使用。它仅由框架使用，允许输入 FlowFile 不经修改传递。如果处理器无法转换其输入（例如因为数据无效），处理器可以将数据路由到 failure 关系。

默认情况下，两种实现也有 success 关系。但是，处理器可以覆盖它定义的关系。它通过实现以下方法来做到这一点：

此方法返回 lynxusapi.relationship.Relationship 对象的列表或集合。如果实现了此方法，success 关系将不会自动可用。如果要使用它，则需要在此列表中创建并返回。无论实现公开哪些关系，failure 和 original 将始终可用。

与 FlowFileTransform 和 RecordTransform 处理器不同，FlowFileSource 处理器默认只有 success 关系。实现可以使用此关系路由创建的 FlowFile。可以通过实现 getRelationships 方法公开额外的关系。对于 FlowFileSource，实现 getRelationships 不会删除 success 关系。getRelationships 返回的任何关系都会出现在 success 关系旁边。

如上所述，ProcessorDetails 和 Java 内部类对处理器很重要。Java 内部类必须在所有处理器上定义，并且必须包含一个名为 implements 的成员，该成员是类实现的 Java 接口的列表。这很重要，因为它允许 Py4J 协议理解如何从 Java 端与此对象交互。

ProcessorDetails 类告诉 Lynxus 有关处理器的信息，以便它可以通过 Lynxus UI 无缝配置处理器。此外，它提供了使用处理器所需的信息的详细信息。ProcessorDetails 类可能有几个不同的成员：

version：处理器的实现版本

description：可在 UI 中显示以说明如何使用处理器的描述。这可以超过一句话，但应保持为几句话或一小段。

tags：与处理器关联的标签或关键字的字符串列表。当用户通过 UI 将处理器添加到 Lynxus 画布时，用户可以搜索关键字以提供可发现性。例如，如果用户搜索 "CSV"，任何名称包含字母 "CSV" 的处理器都会出现。此外，任何具有 "CSV" 标签的处理器也会出现。

dependencies：处理器依赖的 PyPI 依赖项的字符串列表。这些字符串的格式与提供给 pip install 的格式相同。有关更多信息，请参见 添加第三方依赖。

Lynxus 日志记录与任何其他应用程序中的工作方式大致相同，但有一个重要区别。Lynxus 旨在使界面直观且信息丰富，作为该体验的一部分，会显示适当的日志消息。为了适应这一点，处理器不应实例化自己的日志记录器。相反，处理器应简单地使用 self.logger。这将在处理器创建后注入到处理器中。当然，在处理器创建之前它不可用，因此不能从构造函数中访问。但是，它可以在其他任何地方使用。

通常需要创建昂贵的对象并重用它们，而不是创建一次对象、使用它然后丢弃它。为了简化处理，Lynxus 提供了一个名为 onScheduled 的方法。此方法在处理器中可选实现。如果实现了该方法，则定义为：

其中 context 是前面描述的 ProcessContext。该方法没有返回值。每当处理器被调度运行时调用此方法一次（无论是由于用户输入、Lynxus 重启等原因启动）。

类似地，当不再需要资源时通常需要拆除它们。这可以通过实现以下方法来完成：

每当处理器已停止且不再有任何活动任务时调用此方法一次。可以安全地假设调用此方法时不再有任何 transform 方法的调用在运行。

作为 transform()、create()、onScheduled() 和 onStopped() 方法参数的 context 对象通过 getStateManager() 方法提供对状态管理器的访问。状态管理器负责提供用于存储和检索状态的简单 API。有关状态管理器的信息，请参阅 Lynxus 开发者指南。

context.getStateManager() 返回的 Python StateManager 对象提供对底层 Java StateManager 的访问。在 Python StateManager 中，状态以 Python 字典的形式处理（而不是 Java Map）。与 Java 一样，StateManager 可以处理 LOCAL 和 CLUSTER 状态。

如果使用 StateManager 的方法访问状态时发生错误，将抛出 StateException，可以在 Python 处理器的代码中捕获和处理。这使 Python 开发者能够实现处理器，使得当发生状态相关错误时，处理器可以继续运行而不受干扰。如果 StateManager 抛出 StateException 但在处理器代码中未被捕获，则 ProcessSession 将回滚并记录错误。

以下是使用 StateManager 的示例处理器。此示例假设处理器仅在主节点上执行，且仅在一个线程上执行。

lynxusapi.documentation 模块中定义的 @use_case 装饰器可以促进这一点。装饰器接受四个参数：

单个处理器可以有多个 @use_case 装饰器。

在设计和创建处理器时，牢记松耦合的理念很重要。一个处理器不应依赖另一个处理器来执行其任务。话虽如此，构建设计为可以良好协同工作的处理器通常是有利的。例如，能够列出目录中文件的处理器本身可以提供重要的功能。类似地，能够摄取特定文件内容并使该文件内容成为 FlowFile 内容的处理器本身也是重要的功能。但比这些单独功能更强大的是能够组合一个列出目录中所有文件然后将每个文件作为 FlowFile 摄取的流的概念。这是通过结合使用两者来完成的。因此，两个处理器能够以这样一种方式协同工作很重要，即第一个处理器的输出很容易被理解为第二个处理器的输入。

在这种情况下，将这种处理器组合记录为用例是有意义的，以便用户可以理解如何组合这样的管道。这是通过使用 @multi_processor_use_case 装饰器来实现的。此装饰器与 @use_case 装饰器 非常相似，但它不是有一个 configuration 元素，而是有一个 configurations 元素，它是一个 ProcessorConfiguration 对象的 list，其中每个 ProcessorConfiguration 对象都有 processor_type（处理器的名称）和 configuration（说明如何配置该特定处理器）。configuration 元素通常还说明如何连接出站关系。

例如，我们可以这样使用这些装饰器：

请注意，在这种情况下，我们能够明确告诉用户 FetchFile 的 Filename 属性应设置为值 ${absolute.path}/${filename}，因为我们知道 ListFile 处理器将为我们生成这些属性。

如果一个或多个处理器在 Python 包中定义，包应定义一个 requirements.txt 文件，声明包中任何处理器所需的所有第三方依赖。文件结构通常如下所示：

这样，所有依赖将从 requirements.txt 文件为包加载一次。无需为 ProcessorA 加载一次依赖，再为 ProcessorB 加载一次。

如果您的处理器不是 Python 包的一部分，则可以使用 ProcessorDetails 内部类的 dependencies 成员声明其依赖。这是一个字符串列表，指示处理器依赖的 PyPI 模块。格式与 PyPI 期望的格式相同。这为声明第三方依赖提供了便利，而无需将处理器打包到包中。

例如，要指示处理器需要安装 pandas，实现可能如下所示：

但是，通常需要声明依赖的特定版本。也可能需要定义多个依赖。我们可以这样做：

这里，我们接受 pandas 的任何版本（虽然首选最新版本），并且我们需要 numpy 的 1.20.0 版本。

当用户第一次创建给定类型的 Lynxus 处理器时，Lynxus 将为该处理器创建一个单独的 Python 环境 (venv)。它将使用 pip 仅从 PyPI 将指定的依赖安装到该处理器的适当 Python 环境中。因此，一个处理器的依赖不会提供给另一个处理器。

除此之外，一个版本的处理器依赖不会提供给处理器的其他版本。因此，例如，如果我们有两个不同版本的同一处理器可用，版本 0.0.1 和版本 0.0.2，版本 0.0.1 所需的依赖不会提供给版本 0.0.2，除非处理器的版本 0.0.2 也声明了这些依赖。

然而，某些环境不能使用 pip 进行包管理。例如，在隔离网络的环境中，或者有严格安全策略的环境中，pip 可能不可用。在这种情况下，可以使用 Lynxus 归档 (NAR) 格式打包 Python 处理器。这是一个 .zip 文件，具有以下特定布局，并使用 .nar 文件扩展名：