Ufdjrw

I have to process a huge dataframe, download files from a service by the id column of the dataframe. The logic to download, and all the changes are prepared, but I am not sure what is the best way to make a loop around this. I run this on Databricks, which is why I need to perform the processes in chunks.

The dataframe has a "status" column, which can hold the following values:

"todo", "processing", "failed", "succeeded"

In the while loop I want to perform the following tasks:

while (there are rows with status "todo") 
 - get the first 10 rows if status is todo (DONE)
 - start processing the dataframe, update status to processing (DONE)
 - download files (call UDF), update status to succeeded or failed
 (DONE, not in the code here)

I would like to run this until all the rows' status are other then todo! The problem is that this while loop is not finishing, because the dataframe itself is not updated. It needs to be assigned to another dataframe, but then how to add the new one to the loop?

My code right now:

while(statusDoc.where("status == 'todo'").count > 0) 
 val todoDF = test.filter("status == 'todo'")

 val processingDF = todoDF.limit(10).withColumn("status", when(col("status") === "todo", "processing")
 .otherwise(col("status")))

 statusDoc.join(processingDF, Seq("id"), "outer")
 .select($"id", 
 statusDoc("fileUrl"), 
 coalesce(processingDF("status"), statusDoc("status")).alias("status"))

The join should go like this:

val update = statusDoc.join(processingDF, Seq("id"), "outer")
 .select($"id", statusDoc("fileUrl"),
 coalesce(processingDF("status"), statusDoc("status")).alias("status"))

Then this new update dataframe should be used for the next round of loop.

One thing to remember here is that DataFrame (Spark) are not mutable because they are distributed. You have no guarantee that a given modification would be properly propagated across all the network of executors, if you make some. And you also have no guarantee that a given portion of the data has not already been used somewhere else (in another node for example).

One thing you can do though is add another column with the updated values and remove the old column.

val update = statusDoc.
 .withColumnRenamed("status", "status_doc")
 .join(processingDF, Seq("id"), "outer")
 .withColumn("updated_status", udf((stold: String, stold: String) => if (stnew != null) stnew else stold).apply(col("status"), col("status_doc"))
 .drop("status_doc", "status")
 .withColumnRenamed("updated_status", "status")
 .select("id", "fileUrl", "status")

Then make sure you replace "statusDoc" with the "update" DataFrame. Do not forget to make the DataFrame a "var" instead of a "val". I'm surprised your IDE has not yelled yet.

Also, I'm sure you can think of a way of distributing the problem so that you avoid the while loop - I can help you do that but I need a clearer description of you issue. If you use a while loop, you won't use the full capabilities of your cluster because the while loop is only executed on the master. Then, you'll treat only 10 lines at a time, each time. I'm sure you can append all data you need to the whole DataFrame in a single map operation.