[
https://issues.apache.org/jira/browse/SPARK-28125?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hyukjin Kwon updated SPARK-28125:
---------------------------------
Priority: Major (was: Blocker)
> dataframes created by randomSplit have overlapping rows
> -------------------------------------------------------
>
> Key: SPARK-28125
> URL: https://issues.apache.org/jira/browse/SPARK-28125
> Project: Spark
> Issue Type: Bug
> Components: ML, MLlib, PySpark, Spark Core
> Affects Versions: 2.4.3
> Environment: Run with Databricks Runtime 5.3 ML (includes Apache
> Spark 2.4.0, Scala 2.11)
>
> More details on the environment:
> [https://docs.databricks.com/release-notes/runtime/5.3ml.html]
> The python package versions:
> [https://docs.databricks.com/release-notes/runtime/5.3ml.html#python-libraries]
> Reporter: Zachary
> Priority: Major
>
> It appears that the function randomSplit on a DataFrame creates a separate
> execution plan for each of the result DataFrames, or at least that's the
> impression I get from reading a few StackOverflow pages on it:
> [https://stackoverflow.com/questions/38379522/how-does-spark-keep-track-of-the-splits-in-randomsplit/38380023#38380023]
> [https://stackoverflow.com/questions/32933143/how-does-sparks-rdd-randomsplit-actually-split-the-rdd/32933366]
>
> Because of the separate executions, it is easy to create a situation where
> the Dataframes returned by randomSplit have overlapping rows. Thus if people
> are relying on it to split a dataset into training and test, then they could
> easily end up with the same rows in both sets, thus causing a serious problem
> when running model evaluation.
>
> I know that if you call .cache() on the RDD before calling .randomSplit then
> you can be assured that the returned frames have unique rows, but this
> work-around is definitely not obvious. I did not know about this issue and
> ended up creating improper data sets when doing model training and
> evaluation. Something should be adjusted in .randomSplit so that under all
> circumstances, the returned Dataframes will have unique rows.
>
> Here is a Pyspark script I wrote that re-creates the issue and includes the
> work-around line that fixes it as a temporary workaround:
>
> {code:java}
> import numpy as np
> from pyspark.sql import Row
> from pyspark.sql.functions import *
> from pyspark.sql.types import *
> N = 100000
> ratio1 = 0.85
> ratio2 = 0.15
> gen_rand = udf(lambda x: int(np.random.random()*50000 + 2), IntegerType())
> orig_list = list(np.zeros(N))
> rdd = sc.parallelize(orig_list).map(int).map(lambda x: {'ID': x})
> df = sqlContext.createDataFrame(rdd.map(lambda x: Row(**x)))
> dfA = df.withColumn("ID2", gen_rand(df['ID']))
> orig_list = list(np.zeros(N))
> rdd = sc.parallelize(orig_list).map(int).map(lambda x: {'ID': x})
> df = sqlContext.createDataFrame(rdd.map(lambda x: Row(**x)))
> dfA = df.withColumn("ID2", gen_rand(df['ID']))
> dfA = dfA.select("ID2").distinct()
> dfA_els = dfA.rdd.map(lambda x: x['ID2']).collect()
> print("This confirms that if you look at the parent Dataframe, the ID2 col
> has unqiue values")
> print("Num rows parent DF: {}".format(len(dfA_els)))
> print("num unique ID2 vals: {}".format(len(set(dfA_els))))
> #dfA = dfA.cache() #Uncommenting this line does fix the issue
> df1, df2 = dfA.randomSplit([ratio2, ratio1])
> df1_ids = set(df1.rdd.map(lambda x: x['ID2']).distinct().collect())
> df2_ids = set(df2.rdd.map(lambda x: x['ID2']).distinct().collect())
> num_inter = len(df1_ids.intersection(df2_ids))
> print()
> print("Number common IDs between the two splits: {}".format(num_inter))
> print("(should be zero if randomSplit is working as expected)")
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
