Python PySpark: подсчет количества строк по неделям с неделей, начиная с понедельника и заканчивая в воскресенье

Question

У меня есть фрейм данных, который содержит следующие столбцы:

ID  Scheduled Date
241 10/9/2018
423 9/25/2018
126 9/30/2018
123 8/13/2018
132 8/16/2018
143 10/6/2018

Я хочу подсчитать общее количество идентификаторов за неделю.В частности, я хочу, чтобы неделя всегда начиналась в понедельник и всегда заканчивалась в воскресенье.

Я уже достиг этого в Jupyter Notebook:

weekly_count_output = df.resample('W-Mon', on='Scheduled Date', label='left', closed='left').sum().query('count_row > 0') 
weekly_count_output = weekly_count_output.reset_index() 
weekly_count_output = weekly_count_output[['Scheduled Date', 'count_row']] 
weekly_count_output = weekly_count_output.rename(columns = {'count_row': 'Total Count'}) 

Но я не знаю, как написать приведенный выше код в синтаксисе Python PySpark.Я хочу, чтобы мои итоговые результаты выглядели так:

Scheduled Date  Total Count
8/13/2018       2
9/24/2018       2
10/1/2018       1
10/8/2018       1

Обратите внимание, что запланированная дата всегда является понедельником (с указанием начала недели), а общее число отсчитывается с понедельника по воскресенье этой недели.

Answer 1

Спасибо Get Last Monday в Spark за определение функции previous_day .

Первый импорт,

from pyspark.sql.functions import *
from datetime import datetime

Предполагаяваши входные данные как в моем DF (DataFrame)

cols = ['id', 'scheduled_date']
vals = [
            (241, '10/09/2018'),
            (423, '09/25/2018'),
            (126, '09/30/2018'),
            (123, '08/13/2018'),
            (132, '08/16/2018'),
            (143, '10/06/2018')
            ]

df = spark.createDataFrame(vals, cols)

Это функция, определенная

def previous_day(date, dayOfWeek):
    return date_sub(next_day(date, 'monday'), 7)

# Converting the string column to timestamp.
df = df.withColumn('scheduled_date', date_format(unix_timestamp('scheduled_date', 'MM/dd/yyy') \
       .cast('timestamp'), 'yyyy-MM-dd'))

df.show()
+---+--------------+
| id|scheduled_date|
+---+--------------+
|241|    2018-10-09|
|423|    2018-09-25|
|126|    2018-09-30|
|123|    2018-08-13|
|132|    2018-08-16|
|143|    2018-10-06|
+---+--------------+

# Returns the first monday of a week
df_mon = df.withColumn("scheduled_date", previous_day('scheduled_date', 'monday'))

df_mon.show()
+---+--------------+
| id|scheduled_date|
+---+--------------+
|241|    2018-10-08|
|423|    2018-09-24|
|126|    2018-09-24|
|123|    2018-08-13|
|132|    2018-08-13|
|143|    2018-10-01|
+---+--------------+

# You can groupBy and do agg count of 'id'.
df_mon_grp = df_mon.groupBy('scheduled_date').agg(count('id')).orderBy('scheduled_date')

# Reformatting to match your resulting output.
df_mon_grp = df_mon_grp.withColumn('scheduled_date', date_format(unix_timestamp('scheduled_date', "yyyy-MM-dd") \
                       .cast('timestamp'), 'MM/dd/yyyy'))

df_mon_grp.show()
+--------------+---------+
|scheduled_date|count(id)|
+--------------+---------+
|    08/13/2018|        2|
|    09/24/2018|        2|
|    10/01/2018|        1|
|    10/08/2018|        1|
+--------------+---------+