我先看了你的另一个重复问题（后来被标记为重复），所以我会回答这个问题。
是的，您可以像在使用partition by命令的查询中一样使用窗口函数。请参阅How to find longest sequence of consecutive dates?
使用类似的方法，但使用 Dataframe 转换，您可以实现相同的效果。

from pyspark.sql import Window
from pyspark.sql.functions import row_number

lst=[[1,1],[2,1],[3,1],[4,1],[5,0],[6,0],[7,0],[8,1],[9,1],[10,1]]
df=spark.createDataFrame(lst,['ID','Sensor'])

# define the window specification

w=Window.partitionBy(df['Sensor']).orderBy(df['ID'])

group_df=df.select('*',(df['ID']-row_number().over(w)).alias('grp')).orderBy('ID')
count_df=group_df.groupBy('grp').count()

# get result by joining sequence counts df back to df containing original columns

group_df\
  .join(count_df,count_df['grp']==group_df['grp'])\
  .select('ID','Sensor','count')\
  .filter('Sensor=1')\
  .orderBy('ID')\
  .show()

产生所需的序列长度：

+---+------+-----+                                                              
| ID|Sensor|count|
+---+------+-----+
|  1|     1|    4|
|  2|     1|    4|
|  3|     1|    4|
|  4|     1|    4|
|  8|     1|    3|
|  9|     1|    3|
| 10|     1|    3|
+---+------+-----+

您描述的解决方法（collect on nodes）可以通过RDD.mapPartitions来实现。这允许您提供一个跨整个分区Map的函数，这意味着您可以在数据子集中的连续样本之间迭代。注意，您需要确保确定分区何时以1开始或结束，并组合分区之间的序列。这可能有点难看，但应该是可能的。
请注意，您需要先对数据进行sort（如果尚未进行）。