我有几行空格分隔的输入数据:

Name Company Start_Date End_Date 
Naresh HDFC 2017-01-01 2017-03-31
Anoop ICICI 2017-05-01 2017-07-30

我需要输出为:

Naresh HDFC 2017 01
Naresh HDFC 2017 02
Naresh HDFC 2017 03
Anoop ICICI 2017 05
Anoop ICICI 2017 06
Anoop ICICI 2017 07 

我已经为这些数据制作了一个文本文件，并将其放在我的 Hadoop 集群上，我已经编写了代码，但在获取输出时遇到了一些问题。请帮忙。 我不知道如何从条目中提取月份并将它们放入范围函数中，所以我在范围函数中硬编码了值 3。
代码:

from pyspark import SparkConf,SparkContext
from pyspark.sql import SQLContext,Row
from pyspark.sql.types import *
import datetime
sc = SparkContext()
sqlcon = SQLContext(sc)

month_map={'01':1,'02':2,'03':3,'04':4,'05':5,'06':6,'07':7,'08':8,'09':9,
'10':10,'11':11,'12':12}

def get_month(str):
    return datetime.date(int(str[:4]),month_map[str[5:7]],int(str[8:10]))

def parse_line(str):
    match = str.split()
    return (Row(name = match[0],type = match[1],start_date = 
    get_month(match[2]),end_date = get_month(match[3])))


#-----------------create RDD---------------

filepath = '/user/vikasmittal/Innovacer_data.txt'
rdd1 = sc.textFile(filepath)
rdd2 =rdd1.map(parse_line)
for i in range(3):
    rdd3 = rdd2.map(lambda l:(l.name,l.type,l.start_date.year,i))
    print(rdd3.collect())

最佳答案

加载数据后，将其转换为数据框，并使用 to_date 或 cast("日期")

import pyspark.sql.functions as psf
df = sqlcon\
    .createDataFrame(rdd2, ['Name', 'Company', 'Start_Date', 'End_Date'])\
    .withColumn("Start_Date", psf.to_date("Start_Date"))\
    .withColumn("End_Date", psf.to_date("End_Date"))
df.show()
    +------+-------+----------+----------+
    |  Name|Company|Start_Date|  End_Date|
    +------+-------+----------+----------+
    |Naresh|   HDFC|2017-01-01|2017-03-31|
    | Anoop|  ICICI|2017-05-01|2017-07-30|
    +------+-------+----------+----------+

我们将应用 UDF 来计算 Start_Date 和 End_Date 之间的日期范围:

from dateutil.relativedelta import relativedelta
def month_range(d1, d2):
    return [d1 + relativedelta(months=+x) for x in range((d2.year - d1.year)*12 + d2.month - d1.month + 1)]

import pyspark.sql.functions as psf
from pyspark.sql.types import *
month_range_udf = psf.udf(month_range, ArrayType(DateType()))

我们现在可以将它应用于 Start_Date 和 End_Date 并展开数组以每行只获取一个日期:

df = df.withColumn("Date", psf.explode(month_range_udf("Start_Date", "End_Date")))
df.show()

    +------+-------+----------+----------+----------+
    |  Name|Company|Start_Date|  End_Date|      Date|
    +------+-------+----------+----------+----------+
    |Naresh|   HDFC|2017-01-01|2017-03-31|2017-01-01|
    |Naresh|   HDFC|2017-01-01|2017-03-31|2017-02-01|
    |Naresh|   HDFC|2017-01-01|2017-03-31|2017-03-01|
    | Anoop|  ICICI|2017-05-01|2017-07-30|2017-05-01|
    | Anoop|  ICICI|2017-05-01|2017-07-30|2017-06-01|
    | Anoop|  ICICI|2017-05-01|2017-07-30|2017-07-01|
    +------+-------+----------+----------+----------+

我们现在可以从 Date 列中提取 year 和 month:

res = df.select(
    "Name", 
    "Company", 
    psf.year("Date").alias("year"), 
    psf.month("Date").alias("month")
)
res.show()

    +------+-------+----+-----+
    |  Name|Company|year|month|
    +------+-------+----+-----+
    |Naresh|   HDFC|2017|    1|
    |Naresh|   HDFC|2017|    2|
    |Naresh|   HDFC|2017|    3|
    | Anoop|  ICICI|2017|    5|
    | Anoop|  ICICI|2017|    6|
    | Anoop|  ICICI|2017|    7|
    +------+-------+----+-----+

关于hadoop - Apache Spark - Python - 如何在 Pyspark 中使用范围函数，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/46266576/