数据解析和特征工程管道的设计模式

Question

我知道这个问题过去在这些板上被问过几次，但我有一个更通用的版本，这可能也适用于将来其他人的项目。

简而言之 - 我正在构建一个 ML 系统（使用 Python，但在这种情况下语言选择不是很关键），该系统的 ML 模型位于发生的操作管道的末尾：

• 数据上传
• 数据解析
• 特征工程
• 特征工程（与上一步不同的逻辑括号中）
• 特征工程（与之前的步骤处于不同的逻辑括号中）

...（更多步骤，如最后 3 个）

• 数据传递到机器学习模型

上述每个步骤都有其必须采取的一系列操作，以便构建适当的输出，然后将其用作下一个步骤的输入等。这些子步骤依次可以完全解耦彼此之间，或者其中一些可能需要在那个大步骤中执行一些步骤，首先完成，以生成这些后续步骤使用的数据。 现在的问题是，我需要构建一个自定义管道，这将使在混合中添加新步骤（无论大小）变得非常容易，而不会扰乱现有的步骤。

到目前为止，我已经有了从架构角度来看这可能是什么样子的概念，如下所示：

在查看这个架构时，我立即想到了责任链设计模式，它管理大步骤（1、2、...、n），并且每个大步骤都有自己的小版本的责任链发生在他们的内部，这对于 NO_REQ 步骤独立发生，然后对于 REQ 步骤独立发生（REQ 步骤循环，直到全部完成）。通过在大步骤和小步骤内运行逻辑的共享接口，它可能会运行得相当整齐。

但是，我想知道是否有更好的方法？此外，我不喜欢责任链的一点是，它需要一个人添加新的大/小步骤，总是编辑设置步骤包的逻辑的“内容”，手动包含新添加的步骤。我很想构建一些东西，它只是扫描每个大步骤下特定于步骤的文件夹，并自行构建 NO_REQ 和 REQ 步骤的列表（以维护开放/封闭的 SOLID 原则）。

如有任何想法，我将不胜感激。

Answer 1

我最近做了类似的事情。挑战在于允许“步骤”轻松插入而无需进行太多更改。所以，我所做的是，如下所示。核心思想是定义接口“流程”并固定输入和输出格式。

代码：

class Factory:
    def process(self, input):
        raise NotImplementedError

class Extract(Factory):
    def process(self, input):
        print("Extracting...")
        output = {}
        return output

class Parse(Factory):
    def process(self, input):
        print("Parsing...")
        output = {}
        return output

class Load(Factory):
    def process(self, input):
        print("Loading...")
        output = {}
        return output

pipeline = {
    "Extract" : Extract(),
    "Parse" : Parse(),
    "Load" : Load(),
}

input_data = {} #vanilla input
for process_name, process_instance in pipeline.items():
    output = process_instance.process(input_data)
    input_data = output

输出：

Extracting...
Parsing...
Loading...

因此，如果您需要在解析后添加“步骤”、“append_headers”，您需要做的就是，

#Defining a new step.
class AppendHeaders(Factory):
    def process(self, input):
        print("adding headers...")
        output = {}
        return output

pipeline = {
    "Extract" : Extract(),
    "Append headers": AppendHeaders(), #adding a new step
    "Parse" : Parse(),
    "Load" : Load(),
}

新输出：

Extracting...
adding headers...
Parsing...
Loading...

在您的情况下，您可能想要扫描特定文件夹中的 REQ/NOT_REQ 的额外要求，可以添加为 json 中的字段并将其加载到管道中，这意味着，仅当标志设置为 REQ 时才创建这些“步骤”对象。

不确定这个想法能有多大帮助。想，我就传达一下我的想法。

Answer 2

看看汉密尔顿。您可以通过使用先决函数名称作为输入变量来编写函数来编码和连接步骤。类似于以下内容：

# step1.py

def step1(...) -> DF:
    ...


def step2(...) -> DF:
    ...


def step1_star(...) -> DF:
    ...


# step2_star depends on step2
def step2_star(step2: DF, ...) -> DF:
    ...