I am trying to pull the data from a file for a month and then process it.Basically i need to pull data for each month and do some transformations. Since my job runs daily, i want to make use of it and populate data for that month till the run_date.
I have two approaches in mind:
Approach 1:
populate the data for previous month alone. For eg, if my current_date or run_date is in the month of May
, i would be populating the data for the month of April
.
This can be achieved by pulling the month from current_date()
and subtracting 1
from it.
Something similar to below:
df.filter(month(to_date(col("startDate")))===month(to_date(current_date())-1))
This is just an idea. This code wont achieve what i am trying to do since i am subtracting month part alone and not considering Year
part.
But in this case, my job will be running daily to populate same data for the whole month. Doesn't make sense to do it.
Approach 2:
If my current_date is 2020-05-27
, i want to pull the data from 2020-05-01 to 2020-05-26
.
If my current date is 2020-06-01
, it should populate the data for the month of May that is from 2020-05-01 to 2020-05-31
.
I want to implement Approach 2. The only idea i can think of was to write couple of Case
statement to check for dates and accordingly populate it.
Can someone please share some idea on it. Is there any slightly straight forward way.
I am using Spark 1.5
Check if this helps-
val data =
"""
|2018-04-07 07:07:17
|2018-04-07 07:32:27
|2018-04-07 08:36:44
|2018-04-07 08:38:00
|2018-04-07 08:39:29
|2018-04-08 01:43:08
|2018-04-08 01:43:55
|2018-04-09 07:52:31
|2018-04-09 07:52:42
|2019-01-24 11:52:31
|2019-01-24 12:52:42
|2019-01-25 12:52:42
""".stripMargin
val df = spark.read
.schema(StructType(Array(StructField("startDate", DataTypes.TimestampType))))
.csv(data.split(System.lineSeparator()).toSeq.toDS())
df.show(false)
df.printSchema()
Output-
+-------------------+
|startDate |
+-------------------+
|2018-04-07 07:07:17|
|2018-04-07 07:32:27|
|2018-04-07 08:36:44|
|2018-04-07 08:38:00|
|2018-04-07 08:39:29|
|2018-04-08 01:43:08|
|2018-04-08 01:43:55|
|2018-04-09 07:52:31|
|2018-04-09 07:52:42|
|2019-01-24 11:52:31|
|2019-01-24 12:52:42|
|2019-01-25 12:52:42|
+-------------------+
root
|-- startDate: timestamp (nullable = true)
current date
val filterCOl = (currentDate: String) => when(datediff(date_format(lit(currentDate), "yyyy-MM-dd")
,date_format(lit(currentDate), "yyyy-MM-01"))===lit(0),
date_format(col("startDate"), "yyyy-MM") ===
date_format(concat_ws("-",year(lit(currentDate)), month(lit(currentDate)) -1), "yyyy-MM")
).otherwise(to_date(col("startDate"))
.between(date_format(lit(currentDate), "yyyy-MM-01"), lit(currentDate)))
var currentDateStr = "2018-04-08"
df.filter(filterCOl(currentDateStr)).show(false)
Output-
+-------------------+
|startDate |
+-------------------+
|2018-04-07 07:07:17|
|2018-04-07 07:32:27|
|2018-04-07 08:36:44|
|2018-04-07 08:38:00|
|2018-04-07 08:39:29|
|2018-04-08 01:43:08|
|2018-04-08 01:43:55|
+-------------------+
currentDateStr = "2018-05-01"
df.filter(filterCOl(currentDateStr)).show(false)
Output-
+-------------------+
|startDate |
+-------------------+
|2018-04-07 07:07:17|
|2018-04-07 07:32:27|
|2018-04-07 08:36:44|
|2018-04-07 08:38:00|
|2018-04-07 08:39:29|
|2018-04-08 01:43:08|
|2018-04-08 01:43:55|
|2018-04-09 07:52:31|
|2018-04-09 07:52:42|
+-------------------+