如何识别 Spark Dataframe 中的离散状态振荡？

如何解决如何识别 Spark Dataframe 中的离散状态振荡？

用户 U1 在时间 t1、t2、t3 移动通过区域 Z1、Z2、Z3

用户 U1 在 t1、t2、t3、t4 处来回穿过 Z1、Z2 区域

这就是我所说的用户 « OSCILLATING »。

这被认为是一种振荡：用户 U1 从 Z1 到 Z2，然后再到 Z1。用户访问Z1不止一次，尽管他只访问了Z2一次。

Z1 ==> Z2 ==> Z1

用户 U1 在时间 t1、t2、t3、t4 t5 分别从 Z1 到 Z3，然后到 Z2、Z3 和 Z1。

用户在 3 个区域之间摆动。

对于前面的例子，我们认为这种运动是一种振荡，因为用户访问 Z1 和 Z3 的次数超过了一次，尽管他只访问了 Z2 一次。

为了便于计算，我们可以将用户在其中振荡的最大区域数设置为 5 个区域。

我想创建一个跟踪振荡的列。

对于给定的用户，如果他正在振荡，则为行提供相同的振荡 ID。

如果没有振荡，设置为NULL或设置为0

例如：

要复制/粘贴的示例数据：

Zone,time,person,Oscillation_ID
A,1,ABC,1
B,2,1
A,3,4,5,6,1
C,7,2
D,8,2
E,9,2
C,10,11,12,13,14,15,16,17,2
Z,18,3
X,19,4
Y,20,5

因为我正在处理数十亿条记录，所以我需要一个有效的解决方案。

我使用的是 Spark 2.3

我接受 Scala 和 Python (pyspark) 两种解决方案。

解决方法

这是一个使用 Pandas UDF 的窗函数在按人分组后为每一行分配一个振荡 id 的解决方案。

我没有限制振荡中的最大区域，因为这会引发一系列进一步的业务逻辑问题。

我已经将行分组在同一振荡中，直到证明否则，即示例数据集中的最后两行处于同一振荡中。

假设输入数据按时间排序：

@pandas_udf(IntegerType())
def assign_oscillation(zones: pd.Series) -> int:
  current_oscillation_zones = []
  is_oscillation_frozen = False
  id = 1

  for zone in zones.tolist():
    if zone in current_oscillation_zones and not is_oscillation_frozen:
      is_oscillation_frozen = True
    elif zone not in current_oscillation_zones and is_oscillation_frozen:
      id += 1
      is_oscillation_frozen = False
      current_oscillation_zones = [zone]
    elif zone in current_oscillation_zones and is_oscillation_frozen:
      pass
    elif zone not in current_oscillation_zones and not is_oscillation_frozen:
      current_oscillation_zones.append(zone)
  return id

windowSpec = (Window.partitionBy(col('person'))
              .orderBy(col('time'))
              .rangeBetween(-sys.maxsize,0))


df.withColumn('Oscillation_ID',assign_oscillation('Zone').over(windowSpec)).show()

我有 PySpark 3 和 Python 3.8。

PySpark 2 可能不支持窗口函数中的 PandasUDF。这是一个不太优雅的解决方案，使用 PandasUDF 中的生成器和与 PySpark 2.3 兼容的 groupBy：

def oscillation_generator(rows: pd.DataFrame) -> pd.DataFrame:
  current_oscillation = pd.DataFrame(data=None)
  current_oscillation_zones = []
  is_oscillation_frozen = False
  id = 1

  for _,row in rows.iterrows():
    if row['Zone'] in current_oscillation_zones[:-1] and not is_oscillation_frozen:
      is_oscillation_frozen = True
      row['Oscillation_ID'] = id
      current_oscillation = current_oscillation.append(row)
    elif row['Zone'] not in current_oscillation_zones and is_oscillation_frozen:
      yield current_oscillation
      id += 1
      is_oscillation_frozen = False
      row['Oscillation_ID'] = id
      current_oscillation = pd.DataFrame(data=[row])
      current_oscillation_zones = [row['Zone']]
    elif row['Zone'] in current_oscillation_zones and is_oscillation_frozen:
      row['Oscillation_ID'] = id
      current_oscillation = current_oscillation.append(row)
    elif row[
      'Zone'] not in current_oscillation_zones and not is_oscillation_frozen:
      current_oscillation_zones.append(row['Zone'])
      row['Oscillation_ID'] = id
      current_oscillation = current_oscillation.append(row)
  yield current_oscillation

@pandas_udf(StructType(
  [StructField('Zone',StringType()),StructField('time',IntegerType()),StructField('person',StructField('Oscillation_ID',IntegerType())]),PandasUDFType.GROUPED_MAP)
def assign_oscillations(rows: pd.DataFrame) -> pd.DataFrame:
  oscillations = oscillation_generator(rows)
  return pd.concat([oscillation for oscillation in oscillations])


df.groupBy(['person']).apply(assign_oscillations).show()