ArrayIndexOutOfBoundsException when creating dynamic DataFrame on Spark

Question

My goal is to create a dataframe dynamically from columns and values from an external dataframe. This is how the dataframe is created with manual schema and data definitions:

val columnSufix: String = "isNull" 

val data = Seq(Row(
                   details.filter(col("DAY").isNull).count(),
                   details.filter(col("CHANNEL_CATEGORY").isNull).count(),
                   details.filter(col("SOURCE").isNull).count(),
                   details.filter(col("PLATFORM").isNull).count()
              )
            )

val schema: StructType = new StructType()
  .add(s"DAY_$columnSufix", LongType)
  .add(s"CHANNEL_CATEGORY_$columnSufix", LongType)
  .add(s"SOURCE_$columnSufix", LongType)
  .add(s"PLATFORM_$columnSufix", LongType)


val testDf: DataFrame = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)


testDf.show(false)

The output when I execute the above script is the following:

columnSufix: String = isNull
data: Seq[org.apache.spark.sql.Row] = List([0,0,0,83845])
schema: org.apache.spark.sql.types.StructType = StructType(StructField(DAY_isNull,LongType,true),StructField(CHANNEL_CATEGORY_isNull,LongType,true),StructField(SOURCE_isNull,LongType,true),StructField(PLATFORM_isNull,LongType,true))
testDf: org.apache.spark.sql.DataFrame = [DAY_isNull: bigint, CHANNEL_CATEGORY_isNull: bigint ... 2 more fields]

So far so good. The problem is when I use the below script that makes the creation of the DataFrame more dynamic, I get an ArrayIndexOutOfBoundsException error:

val cols = details.columns.toSeq.take(4)
val columnSuffix: String = "ISNULL" 

val data = cols.map(column => Row(details.filter(col(column).isNull).count())).toList

val schema = StructType(cols.map(column => StructField(column + s"_$columnSuffix", LongType)))

val testDf: DataFrame = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)

testDf.show(false)

The output when I compiled the above script is similar as the manual dataFrame creation, with the exception that when I run testDf.show(false), it gives me this error:

Caused by: RuntimeException: Error while encoding: java.lang.ArrayIndexOutOfBoundsException
if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 0, DAY_ISNULL), LongType, false) AS DAY_ISNULL#243076L
if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 1, CHANNEL_CATEGORY_ISNULL), LongType, false) AS CHANNEL_CATEGORY_ISNULL#243077L
if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 2, SOURCE_ISNULL), LongType, false) AS SOURCE_ISNULL#243078L
if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 3, PLATFORM_ISNULL), LongType, false) AS PLATFORM_ISNULL#243079L
Caused by: ArrayIndexOutOfBoundsException:

What's this happening? My feeling is that somehow Spark is not sending the values of details.filter(col(column).isNull).count()) straight to the invocation of createDataFrame.

AminMal · Accepted Answer · 2022-07-20 07:19:35Z

1

The problem is, for each column, you're instantiating a row object:

val data = cols.map(column => Row(details.filter(col(column).isNull).count())).toList

So at the end, you'll have:

Seq(
  Row(0),
  Row(0),
  Row(0),
  Row(83845)
)
// so calling .toDF will expect 1 column name

But you're supplying 4 column names, what you intended to do is this:

Seq(Row(0, 0, 0, 83845)) // notice the single row, with 4 columns instead

What you need to do is:

val data: Seq[Long] = cols.map(column => details.filter(col(column).isNull).count())
// Now you have your values here (Seq(0, 0, 0, 83845))

The tricky part here is to convert Seq to Product (Tuples). This depends on how you want to manage it. You can use shapeless, or reflection to do this. Or if you're sure about the number of columns, I recommend this:

val tupled = data.toList match {
  case a :: b :: c :: d :: Nil => (a, b, c, d)
}

At the end:

Seq(tupled).toDF(cols: _*)

You can also address your issue using spark APIs. I suggested the above approach because you seemed to prefer Scala-ish approach. If you need spark APIs, let me know, I'll update the answer.

Update

So let's continue where we left on data variable instead.

Seq(Row(0), Row(0), Row(0), Row(8723))

We want to retrieve a single row with 4 columns instead of 4 rows with 1 column each.

//                       ++ is union-ing rows,  : _* is passing seq as vararg
val row = Row(data.map(_.toSeq).reduce(_ ++ _): _*)

Now let's create dataframe with the given row:

val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(row)), schema)

edited Jul 20, 2022 at 7:19

answered Jul 18, 2022 at 19:01

AminMal

3,1832 gold badges10 silver badges20 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

ultraInstinct Over a year ago

Thank you very much for your answer! I removed .toList based on your recommandation and it gives this ArrayBuffer([0], [0], [0], [18]) instead. You mentioned that you know how to do it with Spark APIs. Would be possible for you to add it in the answer please?

AminMal Over a year ago

@ultraInstinct you're welcome, but I didn't mention to remove the .toList. Sure, I'll attempt to do it today.

Deku07 · Accepted Answer · 2022-07-18 18:01:10Z

My solution is in pyspark

you can try to do same in scala spark as well.

First reduce is present only in python you can find the same functionality in scala I guess fold does the same .

code goes here


from pyspark.sql import functions as F
from pyspark.sql.types import StructType,StructField,LongType,Row
from functools import reduce
data=[(32609,878,199),
      (32609,878,199),
    (32609,832,199),
    (45470,231,199),
    (45470,231,199),
    (45470,231,199),
    (42482,1001,299),
    (42482,16,291)]
df1=spark.createDataFrame(data_with_null_examples,["DAY","CHANNEL_CATEGORY","SOURCE"])
details=df1.withColumn("PLATFORM",F.lit(None))
dyn_columns_list_org=["DAY","CHANNEL_CATEGORY","SOURCE","PLATFORM"]
dynamic_data = Row(*list(map(lambda x:details.filter(F.col(x).isNull()).count(),dyn_columns_list_org)))
print(dynamic_data)
schema = StructType()
dynamic_schema =  reduce(lambda x,y:x.add(y+"_isNull",data_type=LongType()),dyn_columns_list_org,schema)
print(dynamic_schema)
dyn_df = spark.createDataFrame([dynamic_data],schema=dynamic_schema)
dyn_df.show(10,0)

output :-

    +----------+-----------------------+-------------+---------------+
    |DAY_isNull|CHANNEL_CATEGORY_isNull|SOURCE_isNull|PLATFORM_isNull|
    +----------+-----------------------+-------------+---------------+
    |0         |0                      |0            |8              |
     +----------+-----------------------+-------------+---------------+

Kindly upvote if you like my solution.

Collectives™ on Stack Overflow

ArrayIndexOutOfBoundsException when creating dynamic DataFrame on Spark

2 Answers 2

Update

2 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Update

2 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related