last day (15 days later) » 

3:39 PM
-1
Q: How to serialize a DateTimeFormatter in scala?

Lazar GugletaI am using DateTimeFormatter in scala and now it is giving me trouble with a serialization. I read here: Spark and Not Serializable DateTimeFormatter, that I should create an object and use it as static, but it still does not work. Here is the code: import java.time.{ LocalDate} import java.time...

 
Did you try initializing the DateTimeFormatter within your last map method as was also suggested as an option in the URL you provided?
 
Yes, the same error happens.
 
Can you include that code in your original question?
 
Yes, of course. I edited the question.
 
On which line is the error occurring? Is it when you initialize formatter, or when you initialize parsedDates?
 
3:39 PM
I posted the whole error, but I think the problem is formatter.
 
What environment are you using here?
 
Hey
I am in Spark Shell
I am a new user to Spark and Scala
 
I cannot replicate your error in my own spark shell
So I have been trying to pretend I have the error and fix it
 
I can give you the data
 
That works :)
 
3:41 PM
Great, just a second
I will give you the data from concat rdd
1974,1974-06-22
1954,1954-06-19
1954,1954-06-26
2010,2010-07-07
2006,2006-06-24
2014,2014-07-09
1998,1998-06-22
1994,1994-06-17
2010,2010-06-26
1994,1994-07-16
1990,1990-06-08
1974,1974-06-15
1958,1958-06-08
1954,1954-07-03
1958,1958-06-15
1958,1958-06-24
2002,2002-06-08
2014,2014-06-17
1970,1970-06-02
1994,1994-07-09
1998,1998-07-12
2002,2002-06-02
1986,1986-06-22
1986,1986-06-02
1982,1982-07-05
2006,2006-07-01
1990,1990-06-19
1982,1982-06-19
1982,1982-06-22
2014,2014-06-20
1974,1974-06-26
1970,1970-06-17
I think that should be enough
this is the type:
concat: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[120] at distinct at <console>:49
 
How are you creating this
Just parallelizing the seq?
 
Ok here is the whole thing
val Matches = sc.textFile("matches.csv")
val MatchesheaderAndRows = Matches.map(rdd => rdd.split(","))
val matchiz = MatchesheaderAndRows.map(a => (a{5}))
val matchesss = matchiz.map(x => (x.slice(0,4),x))
 
Are you using spark-shell or spark2-shell
 
version 2.4.3
 
Can you provide the contents of matches.csv
This will be easier
 
3:45 PM
Yes sure
Just let me link it somehow
 
copy/paste is fine
 
I can't download onto my machine here from work
Could you copy/paste it in?
 
mid,hometid,visittid,homescore,visitscore,matchdate,matchtype,sid
1,354,355,5,0,1954-06-16,Group 1,45
2,353,356,1,0,1954-06-16,Group 1,145
3,354,353,1,1,1954-06-19,Group 1,145
4,356,355,3,2,1954-06-19,Group 1,45
5,352,350,4,1,1954-06-17,Group 2,5
6,351,349,9,0,1954-06-17,Group 2,112
7,351,352,8,3,1954-06-20,Group 2,142
8,350,349,7,0,1954-06-20,Group 2,45
9,352,350,7,2,1954-06-23,Group 2,112
10,362,364,2,0,1954-06-16,Group 3,5
11,361,363,1,0,1954-06-16,Group 3,112
12,361,364,5,0,1954-06-19,Group 3,112
Yes sure, sorry
 
What comes after definition of matchesss
 
3:48 PM
val concat = matchesss.map(x => x._1.concat(",".concat(x._2))).distinct
and than the class and everything in ti
it*
 
4:03 PM
Is the data ok?
 
Got pulled away for meeting, just got back
Testing up to concat now
 
oh okay
great :D
 
concat.count
whoops
I have 296 values in concat is this correct
 
Yes, correct
 
Are you also creating this test class in spark shell?
 
4:06 PM
yes
 
That is not necessary in spark shell
 
good to know
 
Is there a specific reason you are creating this? I understand you are new so that might be the reason
 
No, I just saw it online on some other post and though I try it
 
error: No implicit Ordering defined for java.time.LocalDate.
parsedDates.max.getDayOfYear - parsedDates.min.getDayOfYear
 
4:09 PM
import java.time.{ LocalDate}
import java.time.format.DateTimeFormatter
 
I have these
import java.time.LocalDate
import java.time.format.DateTimeFormatter
 
yes
Why is that error
implicit val localDateOrdering: Ordering[LocalDate] = Ordering.by(_.toEpochDay)
Do you have this one?
 
Yes
No errors for me now
when I try to collect on evega there is now an error
mat.DateTimeParseException: Text 'matchdate' could not be parsed at index 0
 
I am removing your header and trying again
 
4:12 PM
ok sure
 
28
30
30
29
19
18
30
24
30
30
This is the first 10 results
I modified the last line in the mapper to be (k, parsedDates.max.getDayOfYear - parsedDates.min.getDayOfYear) instead and now these are the first 10 results
(1982,28)
(2006,30)
(2002,30)
(1986,29)
(1966,19)
(1962,18)
(1994,30)
(1990,30)
(2010,30)
(1974,24)
This is the entirety of the code I used (try restarting spark2-shell and copy/pasting these contents in spark2-shell)
val Matches = sc.textFile("tpt/temp/matches.csv")
val MatchesheaderAndRows = Matches.map(rdd => rdd.split(","))
val matchiz = MatchesheaderAndRows.map(a => (a{5}))
val matchesss = matchiz.map(x => (x.slice(0,4),x))
val concat = matchesss.map(x => x._1.concat(",".concat(x._2))).distinct


import java.time.LocalDate
import java.time.format.DateTimeFormatter
implicit val localDateOrdering: Ordering[LocalDate] = Ordering.by(_.toEpochDay)
val evega = concat.map(_.split(",")).keyBy(_(0)).groupByKey().map{case (k, v) => {
 
This is what I get in postgres
 
imgur is blocked behind my company proxy
 
Damn
ok
1954 18
1962 18
1966 19
1958 21
1970 21
1974 24
1978 24
1982 28
1986 29
1990 30
1994 30
2002 30
2002 30
2006 30
2010 30
2014 31
1998 32
Like this
 
(1982,28)
(2006,30)
(1986,29)
(1966,19)
(2002,30)
(1962,18)
(1994,30)
(1974,24)
(2010,30)
(1990,30)
(1978,24)
(1954,18)
(2014,31)
(1958,21)
(1998,32)
(1970,21)
These are all 21 results
I think they match up with yours.
Did you try copy/pasting my code into a fresh instance of spark2-shell?
 
4:17 PM
Yeah, trying as we speak
How do I get the same error
Must be something with my shell?
scala> val evega = concat.map(_.split(",")).keyBy(_(0)).groupByKey().map{case (k, v) => {
| val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
| val parsedDates = v.map(sa => LocalDate.parse(sa(1), formatter))
| (k, parsedDates.max.getDayOfYear - parsedDates.min.getDayOfYear)
| }}
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
Everything before that executes just fine
I will try resetting the shell
 
What command do you use to open spark shell
Is it spark-shell or spark2-shell
 
spark-shell
 
Use spark2-shell
see if that works
 
ok sure
Are you on windows?
I do not have spark2-shell
Ok, now I created evega
But
java.time.format.DateTimeParseException: Text 'matchdate' could not be parsed at index 0
I have to filter it somehow beforehand
 
Sorry, keep getting pulled away
What you should do is remove the header from the matches.csv file if possible
 
4:31 PM
Sorry for bothering you
 
Not a bother :)
Can you edit the matches.csv file to remove the header, or not?
 
Not
The professor is going to test it with header in it
 
61
Q: How do I skip a header from CSV files in Spark?

Hafiz MujadidSuppose I give three files paths to a Spark context to read and each file has a schema in the first row. How can we skip schema lines from headers? val rdd=sc.textFile("file1,file2,file3") Now, how can we skip header lines from this rdd?

I would follow this here
after you define Matches, do this to Matches
Maybe put it in Matches2. And then MatchesheaderAndRows is now a map based on Matches2
 
scala> evega.foreach(println)
(1982,28)
(1994,30)
(1974,24)
(2006,30)
(2010,30)
(2002,30)
(1990,30)
(1986,29)
(1978,24)
(1966,19)
(1954,18)
(1962,18)
(2014,31)
(1958,21)
(1998,32)
(1970,21)
I did it
You are a savouir
saviour*
savior*****
Thank you so much
 
Sure :)
Thanks for accepting my solution
 
4:38 PM
Of course
:D
 

last day (15 days later) »