apachespark-python-如何在pyspark中使用range函数

zlwx9yxi 于 2021-05-29 发布在 Hadoop

关注(0)|答案(2)|浏览(537)

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

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())

hadoop apache-spark

来源：https://stackoverflow.com/questions/46266576/apache-spark-python-how-to-use-range-function-in-pyspark

2条答案

按热度按时间

lb3vh1jj1#

我们会申请 `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` 以及 `explode` 每行仅获取一个日期的数组：

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|
+------+-------+----------+----------+----------+

我们现在可以提取 `year` 以及 `month` 从 `Date` 列：

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|
+------+-------+----+-----+

赞(0）回复(0）举报 2021-05-29

xuo3flqw2#

你可以用Pypark的 to_date 功能如此处所述。
只需导入pyspark.sql.functions*

>>> df = spark.createDataFrame([('1997-02-28 10:30:00',)], ['t'])
>>> df.select(to_date(df.t).alias('date')).collect()
[Row(date=datetime.date(1997, 2, 28))]

您可以按如下方式提取月份：

>>> df = sqlContext.createDataFrame([('2015-04-08',)], ['a'])
>>> df.select(month('a').alias('month')).collect()
[Row(month=4)]

赞(0）回复(0）举报 2021-05-29

我来回答

apachespark-python-如何在pyspark中使用range函数

2条答案

相关问题

热门标签

最新问答