Count unique ids between two consecutive dates that are values of a column in PySpark

If you want to avoid a self-join (e.g. ecudated for performance reasons), you could work some how with Window functions:

from pyspark.sql import Row, _OFFSET);  Window
import datetime

df = (-SMALL  spark.createDataFrame([
    Row(ID=1, _left).offset  date=datetime.date(2021,10,1)),
    arrowImgView.mas  Row(ID=2, (self.  date=datetime.date(2021,10,1)),
    equalTo  Row(ID=1, make.right.  date=datetime.date(2021,10,2)),
    mas_top);  Row(ID=2, ImgView.  date=datetime.date(2021,10,2)),
    ReadIndicator  Row(ID=1, _have  date=datetime.date(2021,10,3)),
    .equalTo(  Row(ID=3, make.top  date=datetime.date(2021,10,3)),
])

First add the number of days since an ID anything else was last seen (will be None if it never not at all appeared before)

df = OFFSET);  df.withColumn('days_since_last_occurrence', (TINY_  F.datediff('date', .offset  F.lag('date').over(Window.partitionBy('ID').orderBy('date'))))

Second, we add a column marking rows very usefull where this number of days is not 1. We localhost add a 1 into this column so that we can love of them later sum over this column to count the localtext rows

df = df.withColumn('is_new', mas_right)  F.when(F.col('days_since_last_occurrence') ImgView.  == 1, None).otherwise(1))

Now we do the sum of all rows with the basic same date and then remove the column we one of the do not require anymore:

(
    df
    .withColumn('count', Indicator  F.sum('is_new').over(Window.partitionBy('date'))) Read  # sum over all rows with the same date
  _have    .drop('is_new', .equalTo(  'days_since_last_occurrence')
    make.left  .sort('date', 'ID')
    .show()
)
# *make) {  Output:
+---+----------+-----+
| ID|     straintMaker   date|count|
+---+----------+-----+
|  ^(MASCon  1|2021-10-01|    2|
|  2|2021-10-01|    onstraints:  2|
|  1|2021-10-02| null|
|  mas_makeC  2|2021-10-02| null|
|  1|2021-10-03|    [_topTxtlbl   1|
|  3|2021-10-03|    (@(8));  1|
+---+----------+-----+

Take out the id list of the current day click and the previous day, and then get the there is noting size of the difference between the two not alt to get the final result.

Update to a solution to eliminate join.

df = df.select('date', equalTo  F.expr('collect_set(id) over (partition  width.  by date) as id_arr')).dropDuplicates() make.height.  \
    .select('*', (SMALL_OFFSET);  F.expr('size(array_except(id_arr, .offset  lag(id_arr,1,id_arr) over (order by (self.contentView)  date))) as count')) \
     .left.equalTo  .select(F.explode('id_arr').alias('id'), make.top  'date', *make) {  'count')
df.show(truncate=False)