Dynamically Expand Arraytype() Columns in Structured Streaming PySpark

Question

I have the following DataFrame:

root
 |-- sents: array (nullable = false)
 |    |-- element: integer (containsNull = true)
 |-- metadata: array (nullable = true)
 |    |-- element: float (containsNull = true)

+----------+---------------------+
|sents     |metadata             |
+----------+---------------------+
|[1, -1, 0]|[0.4991, 0.5378, 0.0]|
|[-1]      |[0.6281]             |
|[-1]      |[0.463]              |
+----------+---------------------+

I want to expand each array item to its own column DYNAMICALLY so that it may look as follows:

+--------+--------+--------+-----------+-----------+-----------+
|sents[0]|sents[1]|sents[2]|metadata[0]|metadata[1]|metadata[2]|
+--------+--------+--------+-----------+-----------+-----------+
|       1|      -1|       0|     0.4991|     0.5378|        0.0|
|      -1|    null|    null|     0.6281|       null|       null|
|      -1|    null|    null|      0.463|       null|       null|
+--------+--------+--------+-----------+-----------+-----------+

but in structured Streaming there are many limitations as to doing things dynamically:

I tried the following:

numcol = df.withColumn('phrasesNum', F.size('sents')).agg(F.max('phrasesNum')).head()
df = df.select(*[F.col('sents')[i] for i in range(numcol[0])],*[F.col('metadata')[i] for i in range(numcol[0])])

Also:

df_sizes = df.select(F.size('sents').alias('sents'))
df_max = df_sizes.agg(F.max('sents'))
nb_columns = df_max.collect()[0][0]

d = c.select(*[F.map_values(c['metadata'][i]).getItem(0).alias('confidenceIntervals'+"{}".format(j)).cast(DoubleType()) for i,j in enumerate(range(F.size('sents')))],
             *[c['sents'][i].alias('phraseSents'+"{}".format(j)).cast(IntegerType()) for i,j in enumerate(range(nb_columns))])

but I cannot use things like .head(), .collect(), or .take() in Structured Streaming to create the numeric variable that indicate the number of columns to dynamically create. Any ideas??

thanks to all

user_10107389 · Answer

Only way you can do this without collecting to driver node(first,take,collect etc), is if you the know the columns you need or the max size of each array column. Here I assumed both columns had a max size of 3, with columns required 0,1,2.

Also in streaming you cant have different schema(columns) between dataframes.

cols=['0','1','2']

from pyspark.sql import functions as F
df.withColumn("struct1", F.struct(*[F.struct((F.col("sents")[int(x)]).alias('sents[{}]'.format(x))) for x in cols]))\
  .withColumn("struct2", F.struct(*[F.struct((F.col("metadata")[int(x)]).alias('metadata[{}]'.format(x))) for x in cols]))\
  .select(*["struct1.{}.*".format(x) for x in ['col{}'.format((int(x)+1)) for x in cols]],
          *["struct2.{}.*".format(x) for x in ['col{}'.format((int(x)+1)) for x in cols]]).show()

#+--------+--------+--------+-----------+-----------+-----------+
#|sents[0]|sents[1]|sents[2]|metadata[0]|metadata[1]|metadata[2]|
#+--------+--------+--------+-----------+-----------+-----------+
#|       1|      -1|       0|     0.4991|     0.5378|        0.0|
#|      -1|    null|    null|     0.6281|       null|       null|
#|      -1|    null|    null|      0.463|       null|       null|
#+--------+--------+--------+-----------+-----------+-----------+

Dynamically Expand Arraytype() Columns in Structured Streaming PySpark

1 Answers