title: Pyspark: Using collect_list over window() with condition
body: <p>I have the following test data:</p>
  
  <pre><code>import pandas as pd
  import datetime
  
  data = {'date': ['2014-01-01', '2014-01-02', '2014-01-03', '2014-01-04', '2014-01-05', '2014-01-06'],
       'customerid': [2, 2, 2, 3, 4, 3], 'names': ['Andrew', 'Pete', 'Sean', 'Steve', 'Ray', 'Stef'], 'PaymentType': ['OI', 'CC', 'CC', 'OI', 'OI', 'OI']}
  data = pd.DataFrame(data)
  data['date'] = pd.to_datetime(data['date'])
  </code></pre>
  
  <p>The following code gives me the list of names with matching customerid's within a two day time-frame:</p>
  
  <pre><code>import pandas as pd
  import datetime
  
  from pyspark.sql.window import Window
  from pyspark.sql import functions as F
  from pyspark.sql import SparkSession
  
  
  spark = SparkSession.builder \
      .master('local[*]') \
      .config("spark.driver.memory", "500g") \
      .appName('my-pandasToSparkDF-app') \
      .config("spark.ui.showConsoleProgress", "false")\
      .getOrCreate()
  spark.conf.set("spark.sql.execution.arrow.enabled", "true")
  spark.conf.set('spark.sql.execution.arrow.maxRecordsPerBatch', 50000)
  spark.sparkContext.setLogLevel("OFF")
  
  
  data = {'date': ['2014-01-01', '2014-01-02', '2014-01-03', '2014-01-04', '2014-01-05', '2014-01-06'],
       'customerid': [2, 2, 2, 3, 4, 3], 'names': ['Andrew', 'Pete', 'Sean', 'Steve', 'Ray', 'Stef'], 'PaymentType': ['OI', 'CC', 'CC', 'OI', 'OI', 'OI']}
  data = pd.DataFrame(data)
  data['date'] = pd.to_datetime(data['date'])
  spark_data= spark.createDataFrame(data)
  
  win = Window().partitionBy('customerid').orderBy((F.col('date')).cast("long")).rangeBetween(
          -(2*86400), Window.currentRow)
  
  result_frame = spark_data.withColumn("names_array", F.collect_list('names').over(win)).sort(F.col("date").asc())
  
  pd_result_frame = result_frame.toPandas()
  </code></pre>
  
  <p>Data:</p>
  
  <pre><code>&lt;pre&gt;
  date      |customerid|PaymentType|names 
  2014-01-01|2         |OI         |Andrew
  2014-01-02|2         |CC         |Pete 
  2014-01-03|2         |CC         |Sean
  2014-01-04|3         |OI         |Steve
  2014-01-05|4         |OI         |Ray
  2014-01-06|3         |OI         |Stef
  &lt;/pre&gt;
  </code></pre>
  
  <p>Resulting table:</p>
  
  <pre><code>&lt;pre&gt;
  date      |customerid|PaymentType|names_array|
  2014-01-01|2         |OI         |['Andrew']
  2014-01-02|2         |CC         |['Andrew', 'Pete'] 
  2014-01-03|2         |CC         |['Andrew', 'Pete', 'Sean']
  2014-01-04|3         |OI         |['Steve']
  2014-01-05|4         |OI         |['Ray']
  2014-01-06|3         |OI         |['Steve', 'Stef']
  &lt;/pre&gt;
  </code></pre>
  
  <p>Now I would like to introduce a condition for F.collect_list. Only those names should be collected into the lists for which PaymentType == 'OI'.</p>
  
  <p>In the end, the table should look like this:</p>
  
  <pre><code>&lt;pre&gt;
  date      |customerid|PaymentType|names_array|
  2014-01-01|2         |OI         |['Andrew']
  2014-01-02|2         |CC         |['Andrew'] 
  2014-01-03|2         |CC         |['Andrew']
  2014-01-04|3         |OI         |['Steve']
  2014-01-05|4         |OI         |['Ray']
  2014-01-06|3         |OI         |['Steve', 'Stef']
  &lt;/pre&gt;
  </code></pre>
  
  <p>Thank you!</p>