PySparks / Databricks 中两列的图形处理

Question

假设我有以下两列数据框

value_1| value_2
----------------
      1|       2
      2|       3
      4|       5
      6|       5
      4|       6

现在我想将我的所有值聚集到一个新的数据帧中，其中列 ID 包含每个出现的值和列 cluster_ID 表示以某种方式一起出现的所有值的最小值：

ID  | cluster_ID
----------------
   1|          1
   2|          1
   3|          1
   4|          4
   5|          4
   6|          4

请注意，即使值 1 和 3 现在有直接链接，它们仍然聚集在 (1, 2, 3) 集群中，因为它们与值 2 有一个连接。

由于我不知道如何以 Sparks 方式解决此问题，因此我尝试执行以下操作：

首先，我创建了一个包含所有 ID 对的列表：

[[1, 2], [2, 3], [4, 5], [6, 5], [4, 6]]

然后我创建了一个列表列表，其中每个子列表都用这个 for 循环表示集群：

id_pair_list = [[1, 2], [2, 3], [4, 5], [6, 5], [4, 6]]

duplicate_list = []

for e in id_pair_list:
  if not duplicate_list:
    duplicate_list = [e]
  else:
    try:
      index = next(i for i, value in enumerate(duplicate_list) if e[0] in value)
      updated_list = duplicate_list[index] 
      updated_list.append(e[1])       
      duplicate_list[index] = updated_list
    except StopIteration:
      pass
      try:
        index = next(i for i, value in enumerate(duplicate_list) if e[1] in value)
        updated_list = duplicate_list[index]
        updated_list.append(e[0]) 
        duplicate_list[index] = updated_list
      except StopIteration:
        duplicate_list.append(e)   
        
set_duplicate_list = []
for e in duplicate_list:
  set_duplicate_list.append(sorted(list(set(e))))

根据需要，结果如下所示：

[[1, 2, 3], [4, 5, 6]]

在此之后，我创建了这样的新数据框：

id_mapping_df = spark.createDataFrame(
    [[set_duplicate_list]], 
    ['col']
).select(
    F.explode('col').alias('ID')
).withColumn(
    'cluster_id', 
    F.array_min('ID')
).withColumn(
    'ID', 
    F.explode('ID')
)

这给了我最终的结果

...但是...

不幸的是，这只适用于我的小示例数据集。 当我用我更大的真实数据集尝试这个时，我突然遇到了问题，一些值出现在多个集群子列表中，不应该是这种情况。

我猜这已经发生了，因为带有 Sparks 的 for 循环是一种反模式，并且通过在我的 4 个节点上分配工作负载，Sparks 并没有保持我的集群列表的一个恒定状态。

如何以更好的 Sparks 兼容方式解决此问题？

THX & BR 进入数字

Answer 1

在我看来，这更像是一个图形问题，而不是聚类问题。在 Databricks 中，您可以通过将相关的 GraphFrames 库上传到集群来使用 GraphFrames。 connectedComponents 算法计算出这些组。我使用的是 graphframes-0.8.0-spark3.0-s_2.12.jar，这取决于您的 Spark (3.x) 和 Scala 版本 (2.12.x)

这是一个简单的例子：

单元格1

%python
from graphframes import *

# Vertices dataframe
v = sqlContext.createDataFrame((
  ( 1, 2 ), ( 2, 3 ), ( 4, 5 ),
  ( 6, 5 ), ( 4, 6 )
)).toDF("id", "id2")

## Edge dataframe
e = sqlContext.createDataFrame((
  (1, 2, "is linked to"),
  (2, 3, "is linked to"),
  (4, 5, "is linked to"),
  (6, 5, "is linked to"),
  (4, 6, "is linked to") 
)).toDF("src", "dst", "relationship")


## Create the graph frame
g = GraphFrame(v, e)
print(g)

单元格2

%python
## The connected components adds a component id to each 'group'
sc.setCheckpointDir("/tmp/graphframes-example-connected-components")

components = g.connectedComponents() ## doesn't work on Spark 1.4
display(components)

此时，components 数据框将包含您需要的所有信息：

如果需要，您可以进一步操作它，例如将其保存到临时视图并在其上运行一些常规 SQL：

细胞3

%python
components.createOrReplaceTempView("tmp")

细胞 4

%sql
SELECT id, component
FROM tmp
UNION
SELECT id2, component
FROM tmp
ORDER BY 1, 2

SQL 结果：

如果您的数据已经在数据框中，只需一个选择和一个 where 过滤器就可以很容易地从您的原始数据框生成边缘数据框，例如，请参阅此处以获取 recent example。

Answer 2

在我看来，这不像是集群。如果您想使用 Spark 进行集群，您可以使用下面的链接找到一些从哪里开始的想法。

https://spark.apache.org/docs/latest/ml-clustering.html