基于数据集partition/iterator逻辑的管道执行的动态实例

Question

不确定这是否可行，但这是我正在尝试做的：-

我想提取函数的部分（步骤）作为单独的节点（到目前为止还可以），但要注意的是我在步骤之上有一个迭代器，它依赖于数据集上的某些逻辑，即重复在数据集的逻辑分区上进行相同的操作（它们是独立的）。

示例代码

def single_node(list_of_numbers):
   modified_list = [] # to store all output
   for x in list_of_numbers: # iteration logic
      x+=1 # Step 1
      x=str(x) # Step 2
      x+="_suffix" # Step 3
      modified_list.append(x) # append to final output
   return modified_list # return

上下文

在提供的示例中，假设当前我有一个节点执行所有步骤。
因此当前管道有一个节点接受 1 个输入和 returns 1 个输出。
随着步骤的复杂性增加，我想将它们作为单独的节点公开。所以我创建了另一个管道，将这 3 个步骤作为单独的节点并将它们连接在一起。（他们的输入和输出）
但我的总体要求没有改变，我想遍历 list_of_numbers 中的所有值，并且对于这个列表中的每个元素，我想调用这个新管道。最后我想合并所有运行的输出并生成一个输出。

有点类似于基于数据集扩展的动态图（管道的多个动态实例）。

需要考虑的其他要点，

我的输入是单个文件。假设我根据一些定义为节点的逻辑对数据集进行分区。所以这个节点可以有多个输出。（确切的计数完全取决于数据集，这里是列表的大小）
对于数据迭代器节点的每个输出，我需要 "spawn" 一个管道。
最后，合并所有 "spawned" 管道的输出。（此逻辑可以再次在具有多个动态输入的合并节点中定义）。

有办法吗？谢谢！

Answer 1

看起来 PartitionedDataSet 或 IncrementalDataSet 可能对您有用。

它们允许您将相似的数据分成单独的块，由文件确定，并在您认为合适的时候对这些块重复操作。

因此，不是启动包含 y 个节点的 x 个管道，而是包含一个处理 x 个数据块的 y 个节点的管道。

此视频中有关 IncrementalDataSet 的更多信息：https://www.youtube.com/watch?v=v7JSSiYgqpg

# nodes.py

from typing import Any, Dict, Callable

def _dict_mapper(dictionary: Dict[str, Any], fun: Callable):
  # Apply your function to the dictionary mapping
  return {k: fun(v) for k, v in dictionary.items()}

def node_0(list_of_strings: Dict[str, str]):
  return _dict_mapper(list_of_strings, lambda x: int(x))

def node_1(list_of_numbers: Dict[str, int]):
  return _dict_mapper(list_of_numbers, lambda x: x+1)

def node_2(list_of_numbers: Dict[str, int]):
  return _dict_mapper(list_of_numbers, lambda x: str(x))

def node_3(list_of_strings: Dict[str, str]):
  return _dict_mapper(list_of_strings, lambda x: f'{x}_suffix')


# catalog.yml
data:
  type: IncrementalDataSet
  dataset: text.TextDataSet
  path: folder/with/text_files/each/containing/single/number/
  filename_suffix: .txt

# pipeline.py

Pipeline([
  node(node_0, inputs='data', outputs='0'),
  node(node_1, inputs='0', outputs='1'),
  node(node_2, inputs='1', outputs='2'),
  node(node_3, inputs='2', outputs='final_output'),
])

基于数据集partition/iterator逻辑的管道执行的动态实例

Dynamic instance of pipeline execution based on dataset partition/iterator logic

python

kedro

示例代码

上下文

需要考虑的其他要点，