如何迭代python列表以获取映射到不同键的增量日期周期

Question

我有一个有多个 id 的表，每个表都有不同的开始日期。结束日期对所有人都将保持不变，并将从今天的日期开始动态更改为上个月的最后一天。我正在尝试遍历每个 id 和相应的开始日期，以获取从开始日期到结束日期的月份列表，映射回 id。

我当前的表格如下所示：

ID      Start_Date      End_Date 
A       2019-12-15      2020-04-30
B       2020-03-03      2020-04-30

我想要的输出表：

ID      Start_Date      End_Date       ID_period
A       2019-12-15      2020-04-30     201912
A       2019-12-15      2020-04-30     202001
A       2019-12-15      2020-04-30     202002
A       2019-12-15      2020-04-30     202003
A       2019-12-15      2020-04-30     202004
B       2020-03-03      2020-04-30     202003
B       2020-03-03      2020-04-30     202004

我已经尝试了以下代码，其中一些更改来自Generate list of months between interval in python

from datetime import datetime, timedelta
from collections import OrderedDict

dates = ["2014-10-10","2016-01-01"]

def monthlist_fast(dates):
    for val in enumerate(dates):
        start = val
        end = dt.date.today().replace(day=1) - timedelta(days=1)
        start, end = [datetime.strptime(_, "%Y-%m-%d") for _ in dates]
        total_months = lambda dt: dt.month + 12 * dt.year
        mlist = []
        for tot_m in range(total_months(start)-1, total_months(end)):
            y, m = divmod(tot_m, 12)
            mlist.append(datetime(y, m+1, 1).strftime("%Y%m"))
        return mlist

我的结果：

['201410',
 '201411',
 '201412',
 '201501',
 '201502',
 '201503',
 '201504',
 '201505',
 '201506',
 '201507',
 '201508',
 '201509',
 '201510',
 '201511',
 '201512',
 '201601']

但我无法找到将这些映射回我的 Id 的方法，尤其是因为我的 start_dates 不断随着不同的 Id 而变化。任何帮助，将不胜感激。谢谢。

Answer 1

对于未来，建议更新到 spark 2.4+，因为 sequence 功能会改变游戏规则。

对于Spark2.1+：

从这个答案修改：Generating monthly timestamps between two dates in pyspark dataframe

from pyspark.sql import functions as F
df.withColumn("monthsDiff", F.months_between("End_Date", "Start_Date"))\
    .withColumn("repeat", F.expr("split(repeat(',', monthsDiff), ',')"))\
     .select("*", F.posexplode("repeat").alias("date", "val"))\
    .withColumn("Id_period", F.expr("""date_format(add_months(Start_Date, date),'yyyyMM')"""))\
    .drop("repeat","val","monthsDiff","date").show()

#+---+----------+----------+---------+
#| ID|Start_Date|  End_Date|Id_period|
#+---+----------+----------+---------+
#|  A|2019-12-15|2020-04-30|   201912|
#|  A|2019-12-15|2020-04-30|   202001|
#|  A|2019-12-15|2020-04-30|   202002|
#|  A|2019-12-15|2020-04-30|   202003|
#|  A|2019-12-15|2020-04-30|   202004|
#|  B|2020-03-03|2020-04-30|   202003|
#|  B|2020-03-03|2020-04-30|   202004|
#+---+----------+----------+---------+

对于Spark2.4+:

from pyspark.sql import functions as F

df.withColumn("Id_period", F.explode(F.expr("""transform(sequence(to_date(start_date),to_date(end_date)\
                                                         ,interval 1 month),x-> date_format(x,'yyyyMM'))"""))).show()

#+---+----------+----------+---------+
#| ID|Start_Date|  End_Date|Id_period|
#+---+----------+----------+---------+
#|  A|2019-12-15|2020-04-30|   201912|
#|  A|2019-12-15|2020-04-30|   202001|
#|  A|2019-12-15|2020-04-30|   202002|
#|  A|2019-12-15|2020-04-30|   202003|
#|  A|2019-12-15|2020-04-30|   202004|
#|  B|2020-03-03|2020-04-30|   202003|
#|  B|2020-03-03|2020-04-30|   202004|
#+---+----------+----------+---------+

Answer 2

假设您正在从文件（我称之为 input.txt）中读取数据，您可以尝试类似

from datetime import datetime

def read_log_file():
    data = []

    with open("input.txt", "r") as input_data:
        input_data.readline()
        for line in input_data:
            data.append(line.strip().split())

    with open("output.txt", "w") as output:
        print("ID\tStart_Date\tEnd_Date\tID_period", file=output)
        for entry_id, start_date, end_date in data:
            end_datetime = datetime.strptime(end_date, "%Y-%m-%d")
            id_period = datetime.strptime(start_date, "%Y-%m-%d")
            while id_period.year < end_datetime.year or id_period.month <= end_datetime.month:
                print("\t".join([entry_id, start_date, end_date, id_period.strftime("%Y%m")]), file=output)
                next_year = id_period.year
                next_month = id_period.month + 1
                if next_month > 12:
                    next_month = 1
                    next_year += 1
                id_period = datetime(year=next_year, month=next_month, day=id_period.day)

产生

ID  Start_Date  End_Date    ID_period
A   2019-12-15  2020-04-30  201912
A   2019-12-15  2020-04-30  202001
A   2019-12-15  2020-04-30  202002
A   2019-12-15  2020-04-30  202003
A   2019-12-15  2020-04-30  202004
B   2020-03-03  2020-04-30  202003
B   2020-03-03  2020-04-30  202004

Answer 3

如果您尝试使用 pyspark 来实现，那么您可以使用内置函数，这也将提供更好的性能。

sequence 和 interval 1 month 将扩展 Start_date 和 End_date 一个月的差距，expr 帮助您运行 sql 函数

import pyspark.sql.functions as f

df1 = df.withColumn('months', f.expr('sequence(to_date(Start_Date), to_date(End_Date), interval 1 month)'))\
    .withColumn('month', f.explode('months'))\
    .withColumn('ID_period', f.date_format('month', 'yyyyMM')).drop('months', 'month')

df1.show()

+---+----------+----------+---------+
| ID|Start_Date|  End_Date|ID_period|
+---+----------+----------+---------+
|  A|2019-12-15|2020-04-30|   201912|
|  A|2019-12-15|2020-04-30|   202001|
|  A|2019-12-15|2020-04-30|   202002|
|  A|2019-12-15|2020-04-30|   202003|
|  A|2019-12-15|2020-04-30|   202004|
|  B|2020-03-03|2020-04-30|   202003|
|  B|2020-03-03|2020-04-30|   202004|
+---+----------+----------+---------+

Answer 4

让我知道它是如何工作的。

import pyspark.sql.functions as f
data = spark.createDataFrame([('A', '2019-12-15', '2020-04-30'), ('B', '2020-03-03', '2020-04-30'), ('C', '2020-04-29', '2020-04-30')], ['ID', 'Start_Date', 'End_Date'])
data = (data
        .withColumn('dateDifferenceArray', f.sequence(f.lit(0), f.datediff(f.col('End_Date'), f.col('Start_Date'))))
        .withColumn('ID_period', f.explode(f.array_distinct(f.expr('transform(dateDifferenceArray, element -> date_format(date_add(Start_Date, element), "yyyyMM"))'))))
        .drop('dateDifferenceArray')
       )
data.show()