首先，我们创建一个列，显示 TransactionId_From 和 TransactionId_To 之间的最大值：

from pyspark.sql.functions import greatest, col

df = df.withColumn(
   'TransactionGreatestID',
   greatest(col('TransactionId_From'), col('TransactionId_To')
                  )

然后，您可以使用dense_rank函数根据 TransactionGreatestID 分配组：

from pyspark.sql.window import *
from pyspark.sql.functions import dense_rank

window = Window.orderBy('TransactionGreatestID')
df = df.withColumn('GroupId', dense_rank().over(window))

根据我的经验，这应该不会花太长的时间来运行。但是，对于上面的解决方案，我们假设数据结构是好的，并且给定 GroupId 的每一行都具有相同的 TransactionGreatestID 值。

使用networkx：

import networkx as nx

G = nx.from_pandas_edgelist(df[['TransactionId_From', 'TransactionId_To']].fillna('nan'),
                            source='TransactionId_From', target='TransactionId_To',
                            create_using=nx.DiGraph)
G.remove_node('nan')

groups = {n: i for i, s in
          enumerate(nx.weakly_connected_components(G), start=1)
          for n in s}

df['GroupId'] = df['TransactionId_From'].map(groups)

输出：

TransactionID  Amount  Barcode  TransactionId_From  TransactionId_To  GroupId
0            120    55.0    10001                 120             133.0        1
1            133    55.0    10001                 120             133.0        1
2            133    31.0    10008                 133               NaN        1
3            135    20.0    10022                 135             140.0        2
4             10    20.0    10022                 135             140.0        2
5            140     8.0    10056                 140             149.0        2
6            140    17.0    10067                 149               NaN        2
7            149     8.0    10056                 140             149.0        2
8            151     5.0    10056                 151               NaN        3