If you can reproduce the following with a unit test, I suggest you open a JIRA.
Thanks > On Mar 4, 2016, at 4:01 AM, James Hammerton <[email protected]> wrote: > > Hi, > > I've come across some strange behaviour with Spark 1.6.0. > > In the code below, the filtering by "eventName" only seems to work if I > called .cache on the resulting DataFrame. > > If I don't do this, the code crashes inside the UDF because it processes an > event that the filter should get rid off. > > Any ideas why this might be the case? > > The code is as follows: >> val df = sqlContext.read.parquet(inputPath) >> val filtered = df.filter(df("eventName").equalTo(Created)) >> val extracted = extractEmailReferences(sqlContext, filtered.cache) // >> Caching seems to be required for the filter to work >> extracted.write.parquet(outputPath) > > where extractEmailReferences does this: >> >> def extractEmailReferences(sqlContext: SQLContext, df: DataFrame): DataFrame >> = { >> val extracted = df.select(df(EventFieldNames.ObjectId), >> extractReferencesUDF(df(EventFieldNames.EventJson), >> df(EventFieldNames.ObjectId), df(EventFieldNames.UserId)) as "references") >> >> extracted.filter(extracted("references").notEqual("UNKNOWN")) >> } > > and extractReferencesUDF: >> def extractReferencesUDF = udf(extractReferences(_: String, _: String, _: >> String)) >> def extractReferences(eventJson: String, objectId: String, userId: String): >> String = { >> import org.json4s.jackson.Serialization >> import org.json4s.NoTypeHints >> implicit val formats = Serialization.formats(NoTypeHints) >> >> val created = Serialization.read[GMailMessage.Created](eventJson) // >> This is where the code crashes if the .cache isn't called > > Regards, > > James
