Fill null values in a row with frequency of other column

Question

In a spark structured streaming context, I have this dataframe :

+------+----------+---------+
|brand |Timestamp |frequency|
+------+----------+---------+
|BR1   |1632899456|4        |
|BR1   |1632901256|4        |
|BR300 |1632901796|null     |
|BR300 |1632899155|null     |
|BR90  |1632901743|1        |
|BR1   |1632899933|4        |
|BR1   |1632899756|4        |
|BR22  |1632900776|null     |
|BR22  |1632900176|null     |
+------+----------+---------+

I would like to replace the null values by the frequency of the brand in the batch, in order to obtain a dataframe like this one :

+------+----------+---------+
|brand |Timestamp |frequency|
+------+----------+---------+
|BR1   |1632899456|4        |
|BR1   |1632901256|4        |
|BR300 |1632901796|2        | 
|BR300 |1632899155|2        |
|BR90  |1632901743|1        |
|BR1   |1632899933|4        |
|BR1   |1632899756|4        |
|BR22  |1632900776|2        |
|BR22  |1632900176|2        |
+------+----------+---------+

I am using Spark version 2.4.3 and SQLContext, with scala language.

Answer 1

With "count" over window function:

val df = Seq(
  ("BR1", 1632899456, Some(4)),
  ("BR1", 1632901256, Some(4)),
  ("BR300", 1632901796, None),
  ("BR300", 1632899155, None),
  ("BR90", 1632901743, Some(1)),
  ("BR1", 1632899933, Some(4)),
  ("BR1", 1632899756, Some(4)),
  ("BR22", 1632900776, None),
  ("BR22", 1632900176, None)
).toDF("brand", "Timestamp", "frequency")

val brandWindow = Window.partitionBy("brand")
val result = df.withColumn("frequency", when($"frequency".isNotNull, $"frequency").otherwise(count($"brand").over(brandWindow)))

Result:

+-----+----------+---------+
|BR1  |1632899456|4        |
|BR1  |1632901256|4        |
|BR1  |1632899933|4        |
|BR1  |1632899756|4        |
|BR22 |1632900776|2        |
|BR22 |1632900176|2        |
|BR300|1632901796|2        |
|BR300|1632899155|2        |
|BR90 |1632901743|1        |
+-----+----------+---------+

Solution with GroupBy:

val countDF = df.select("brand").groupBy("brand").count()


df.alias("df")
  .join(countDF.alias("cnt"), Seq("brand"))
  .withColumn("frequency", when($"df.frequency".isNotNull, $"df.frequency").otherwise($"cnt.count"))
  .select("df.brand", "df.Timestamp", "frequency")

Answer 2

Hi bro I'm a java programmer . It's better to make a loop through the freq column and search for first null and its related brand . so count the number of that till the end of the table and correct the null value of that brand and go for the other null brand and correct it . here is my java solution :(I didn't test this code just wrote it text editor but I hope works well, 70%;)

    //this is your table  +  dimensions
    table[9][3];    
    int repeatCounter = 0;
    String brand;
    boolean thereIsNull = true;
    //define an array to save the address of the specified null brand
    int[tablecolumns.length()] brandmemory; 
    while (thereisnull) {
        for (int i = 0; i < tablecolumns.length(); i++) {
            
            if (array[i][3] == null) {
                 thereIsNull = true;
                brand = array[i][1];
                for (int n = i; n < tablecolumns.length(); i++) {
                    if (brand == array[i][1]) {
                        repeatCounter++;
                         // making an array to save address of  the null brand in table:
                        brandmemory[repeatCounter] = i;
                        else{
                            break ;
                        }
                    }
                    for (int p = 1; p = repeatCounter ; p++) {
                        //changing null values to number of repeats 
                        array[brandmemory[p]][3] = repeatCounter;
                    }
                }
            }
            else{
                continue;
                //check if the table has any null content if no :end of program. 
                for(int w>i ; w=tablecolumns.length();w++ ){
                    if(array[w] != null  ){
                        thereIsNull = false;
                        else{ thereIsNull = true;
                        break;
                        
                    }
                }
            }
        }
    }