如何创建循环以使用 pyspark 迭代连接

Question

我正在研究 Databricks，我有一个包含 BOM（物料清单）列表的数据框：数据框的结构在下面的示例中报告，其中标识符是“父亲”产品的代码（成品）和组件是“子”产品的代码（可以是半成品或原材料）。如果组件是半成品，您也可以在标识符列表中找到它及其组件（但我在数据框中没有任何指示告诉我产品的类型，是成品、半成品还是生的）。

  identifier   component
  xxxx         yyyy
  xxxx         zzzz
  xxxx         aaaa
  aaaa         bbbb
  aaaa         cccc
  bbbb         dddd
  bbbb         eeee
  cccc         ffff
  cccc         mmmm
  ffff         aaaa
  ffff         gggg
  ffff         hhhh
  hhhh         iiii
  hhhh         jjjj

在上面的示例中，有两个最终产品（xxxx 和 ffff）。 xxxx 有 yyyy、zzzz 和 aaaa 作为组件。 aaaa 应该是半成品，因为它也列在标识符列中，由 cccc 和 dddd 组成。 cccc和dddd也是半成品，因为它们列在标识符列，由dddd、eeee、ffff和mmmm组成（应该是原材料，因为它们不在标识符列）。第二个最终产品ffff由aaaa（半成品也用于xxxx），gggg（原材料）和hhhh（半成品，由iiii和jjjj组成，原材料）组成。 我必须按照企业提供的产品列表过滤此数据框，该列表仅包含最终产品：假设我必须仅选择 xxxx（ffff 不在列表中）。问题是，如果我过滤 xxxx，我会丢失与半成品相关的信息（如果我过滤仅选择标识符 xxxx 的数据框，我会获得 3 行，但我要做的是找到一种方法来保持 aaaa， bbbb 和 cccc，以及它们每个组件的详细信息）。所以最终过滤的数据帧应该是

    identifier   component
  xxxx         yyyy
  xxxx         zzzz
  xxxx         aaaa
  aaaa         bbbb
  aaaa         cccc
  bbbb         dddd
  bbbb         eeee
  cccc         ffff
  cccc         mmmm

我试图弄清楚如何通过循环解决这个问题（实际的数据框显然更大，大约 13000 行）但我找不到一个好的起点（我不是 python 专家）。有没有人作为一些好的建议、文档或 sn-p 用作起点？

Answer 1

如果我过滤数据框，只选择标识符 xxxx，我得到 3 行，但我要做的是找到一种方法来保持 aaaa，bbbb 和cccc

你不是说，yyyy、zzzz和aaaa，因为这些都是xxxx所依赖的产品吗？清除这一点，并详细说明您如何获得最终过滤的数据帧（我得到了不同的结果），我很乐意提供帮助。

我仍然很乐意提供帮助，但我仍然不明白您是如何获得最终过滤数据帧的。

因此您选择xxxx 并接收三个关联的组件yyyy、zzzz 和aaaa。然后呢？您获得了这三个组件的关联组件，然后递归地执行此操作，直到获得基础材料？不过，这并不能反映最终过滤数据框中的内容。

编辑如果我正确理解了这个问题，这就是我想出的（我不经常使用熊猫，所以如果有熊猫功能可以使我不会感到惊讶这个可爱）：

def main():

    import pandas as pd
    from queue import Queue

    df = pd.read_csv("data.csv", names=["identifier", "component"])

    result_df = pd.DataFrame()

    selected_identifier = "xxxx"
    identifier_queue = Queue()
    identifier_queue.put(selected_identifier)
    previously_seen_identifiers = set()

    while not identifier_queue.empty():
        current_identifier = identifier_queue.get()
        if current_identifier in previously_seen_identifiers:
            continue
        previously_seen_identifiers.add(current_identifier)
        current_df = df.loc[df["identifier"] == current_identifier]
        result_df = result_df.append(current_df)
        components = current_df[["component"]]["component"].tolist()
        for component in components:
            identifier_queue.put(component)

    print(result_df)

    return 0


if __name__ == "__main__":
    import sys
    sys.exit(main())

它是这样工作的：

• 创建一个标识符队列（也可以是堆栈），这些标识符仍然 需要处理。最初要处理的唯一标识符 我们选择的最终产品（在本例中为xxxx）。
• 当队列不为空时，获取下一个标识符并将其移除 从队列中，从原始数据帧创建一个数据帧 仅包含当前标识符所在的组件 取决于并将此子数据帧附加到result_df 数据框。在我们开始下一次迭代之前，我们采用相关的 当前标识符所依赖的组件并将它们添加到 我们的未处理标识符队列。
• 在循环之前，我们还创建了一个set() 来跟踪所有的 到目前为止我们看到的标识符。在循环中，如果当前标识符 已经看过了，我们忽略它并继续下一个 标识符。这是因为一些产品似乎 有循环依赖。

让我知道你是不是这么想的。