DataFrame-ified zipWithIndex

I am trying to solve the age-old problem of adding a sequence number to a data set. I am working with DataFrames, and there appears to be no DataFrame equivalent to RDD.zipWithIndex. On the other hand, the following works more or less the way I want it to:

val origDF = sqlContext.load(...)    

val seqDF= sqlContext.createDataFrame( => Row.fromSeq(Seq(ln._2) ++ ln._1.toSeq)),
    StructType(Array(StructField("seq", LongType, false)) ++ origDF.schema.fields)

In my actual application, origDF won't be loaded directly out of a file -- it is going to be created by joining 2-3 other DataFrames together and will contain upwards of 100 million rows.

Is there a better way to do this? What can I do to optimize it?


  • Since Spark 1.6 there is a function called monotonically_increasing_id()
    It generates a new column with unique 64-bit monotonic index for each row
    But it isn't consequential, each partition starts a new range, so we must calculate each partition offset before using it.
    Trying to provide an "rdd-free" solution, I ended up with some collect(), but it only collects offsets, one value per partition, so it will not cause OOM

    def zipWithIndex(df: DataFrame, offset: Long = 1, indexName: String = "index") = {
        val dfWithPartitionId = df.withColumn("partition_id", spark_partition_id()).withColumn("inc_id", monotonically_increasing_id())
        val partitionOffsets = dfWithPartitionId
            .agg(count(lit(1)) as "cnt", first("inc_id") as "inc_id")
            .select(sum("cnt").over(Window.orderBy("partition_id")) - col("cnt") - col("inc_id") + lit(offset) as "cnt" )
            .withColumn("partition_offset", udf((partitionId: Int) => partitionOffsets(partitionId), LongType)(col("partition_id")))
            .withColumn(indexName, col("partition_offset") + col("inc_id"))
            .drop("partition_id", "partition_offset", "inc_id")

    This solution doesn't repack the original rows and doesn't repartition the original huge dataframe, so it is quite fast in real world: 200GB of CSV data (43 million rows with 150 columns) read, indexed and packed to parquet in 2 minutes on 240 cores
    After testing my solution, I have run Kirk Broadhurst's solution and it was 20 seconds slower
    You may want or not want to use dfWithPartitionId.cache(), depends on task