How do I add an persistent column of row ids to Spark DataFrame?

by

Spark 2.0

  • This is issue has been resolved in Spark 2.0 with SPARK-14241.

  • Another similar issue has been resolved in Spark 2.1 with SPARK-14393

Spark 1.x

Problem you experience is rather subtle but can be reduced to a simple fact monotonically_increasing_id is an extremely ugly function. It is clearly not pure and its value depends on something that is completely out your control.

It doesn’t take any parameters so from an optimizer perspective it doesn’t matter when it is called and can be pushed after all other operations. Hence the behavior you see.

If you take look at the code you’ll find out this is explicitly marked by extending MonotonicallyIncreasingID expression with Nondeterministic.


I don’t think there is any elegant solution but one way you can handle this is to add an artificial dependency on the filtered value. For example with an UDF like this:

from pyspark.sql.types import LongType
from pyspark.sql.functions import udf

bound = udf(lambda _, v: v, LongType()) 

(df
  .withColumn("rn", monotonically_increasing_id())
  # Due to nondeterministic behavior it has to be a separate step
  .withColumn("rn", bound("P", "rn"))  
  .where("P"))

In general it could be cleaner to add indices using zipWithIndex on a RDD and then convert it back to a DataFrame.

* Workaround shown above is no longer a valid solution (nor required) in Spark 2.x where Python UDFs are subject of the execution plan optimizations.

More Related Contents:

  1. Spark SQL: apply aggregate functions to a list of columns
  2. Multiple Aggregate operations on the same column of a spark dataframe
  3. DataFrame join optimization – Broadcast Hash Join
  4. How to access element of a VectorUDT column in a Spark DataFrame?
  5. What is the maximum size for a broadcast object in Spark?
  6. How to loop through each row of dataFrame in pyspark
  7. Why does Spark think this is a cross / Cartesian join
  8. Convert null values to empty array in Spark DataFrame
  9. How to exclude multiple columns in Spark dataframe in Python
  10. Why is predicate pushdown not used in typed Dataset API (vs untyped DataFrame API)?
  11. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  12. Concatenate columns in Apache Spark DataFrame
  13. How to define partitioning of DataFrame?
  14. How do I add a new column to a Spark DataFrame (using PySpark)?
  15. Spark Dataframe validating column names for parquet writes
  16. Spark: subtract two DataFrames
  17. Dropping a nested column from Spark DataFrame
  18. Updating a dataframe column in spark
  19. Pyspark : forward fill with last observation for a DataFrame
  20. Spark Dataframes UPSERT to Postgres Table
  21. Create new Dataframe with empty/null field values
  22. multiple conditions for filter in spark data frames
  23. Pyspark filter dataframe by columns of another dataframe
  24. Spark SQL broadcast hash join
  25. What is the difference between Apache Spark SQLContext vs HiveContext?
  26. What should be the optimal value for spark.sql.shuffle.partitions or how do we increase partitions when using Spark SQL?
  27. Spark lists all leaf node even in partitioned data
  28. PySpark: How to fillna values in dataframe for specific columns?
  29. How to calculate Median in spark sqlContext for column of data type double
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment