Значения столбцов для динамического определения структуры

Question

У меня есть два вложенных массива, один из которых является строкой, а другой - плавающим. Я хотел бы существенно сжать это и иметь один (значение, переменная) комбо на строку. Я пытался сделать это с помощью всего лишь набора данных, и мне не приходилось прибегать к rdds или udfs, думая, что это будет чище и быстрее.

Я могу превратить массив значений, переменных для каждой строки в структуру значения, переменной, 1 на строку, но, поскольку размеры моего массива различаются, мне приходится использовать моё понимание массива в разных диапазонах. Поэтому я подумал, что могу просто указать длину в столбце и использовать ее. Но поскольку я буду использовать столбец, это синтаксическая ошибка. Любые предложения о том, как использовать столбец для динамического построения такой структуры (без rdd / udf, если это возможно)?:

from pyspark.sql.functions import col, array, struct, explode

DF1 = spark.createDataFrame([(["a", "b", "c", "d", "e", "f"], [1,2,3,4,5,6], 6),
                             (["g"], [7], 1),
                             (["a", "b", "g", "c"], [4,5,3,6], 4),
                             (["c", "d"], [2,3], 2),
                             (["a", "b", "c"], [5,7,2], 3)],
                            ["vars", "vals", "num_elements"])
DF1.show()

arrayofstructs = array(*[struct(
  DF1.vars[c].alias("variables"),
  DF1.vals[c].alias("values")
#) for c in DF1.num_elements]) # <- DOES NOT WORK
) for c in range(10)])         # <- FIXED SIZE DOES WORK

DF2 = DF1.withColumn("new", explode(arrayofstructs))
DF2.show()

DF3 = DF2.filter(DF2.new.variables.isNotNull())
DF3.show()


+------------------+------------------+------------+
|              vars|              vals|num_elements|
+------------------+------------------+------------+
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|
|               [g]|               [7]|           1|
|      [a, b, g, c]|      [4, 5, 3, 6]|           4|
|            [c, d]|            [2, 3]|           2|
|         [a, b, c]|         [5, 7, 2]|           3|
+------------------+------------------+------------+

+------------------+------------------+------------+------+
|              vars|              vals|num_elements|   new|
+------------------+------------------+------------+------+
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[a, 1]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[b, 2]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[c, 3]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[d, 4]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[e, 5]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[f, 6]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|   [,]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|   [,]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|   [,]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|   [,]|
|               [g]|               [7]|           1|[g, 7]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
|               [g]|               [7]|           1|   [,]|
+------------------+------------------+------------+------+
only showing top 20 rows

+------------------+------------------+------------+------+
|              vars|              vals|num_elements|   new|
+------------------+------------------+------------+------+
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[a, 1]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[b, 2]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[c, 3]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[d, 4]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[e, 5]|
|[a, b, c, d, e, f]|[1, 2, 3, 4, 5, 6]|           6|[f, 6]|
|               [g]|               [7]|           1|[g, 7]|
|      [a, b, g, c]|      [4, 5, 3, 6]|           4|[a, 4]|
|      [a, b, g, c]|      [4, 5, 3, 6]|           4|[b, 5]|
|      [a, b, g, c]|      [4, 5, 3, 6]|           4|[g, 3]|
|      [a, b, g, c]|      [4, 5, 3, 6]|           4|[c, 6]|
|            [c, d]|            [2, 3]|           2|[c, 2]|
|            [c, d]|            [2, 3]|           2|[d, 3]|
|         [a, b, c]|         [5, 7, 2]|           3|[a, 5]|
|         [a, b, c]|         [5, 7, 2]|           3|[b, 7]|
|         [a, b, c]|         [5, 7, 2]|           3|[c, 2]|
+------------------+------------------+------------+------+

Answer 1

Вы можете попытаться взломать что-то вроде этого:

from pyspark.sql.functions import col, lit, posexplode, expr, split

(DF1
    .select("*", posexplode(split(expr("repeat('_', num_elements - 1)"), '_')))
    .select(col("vars").getItem(col("pos")),col("vals").getItem(col("pos")))
    .show())

# +---------+---------+
# |vars[pos]|vals[pos]|
# +---------+---------+
# |        a|        1|
# |        b|        2|
# |        c|        3|
# |        d|        4|
# |        e|        5|
# |        f|        6|
# |        g|        7|
# |        a|        4|
# |        b|        5|
# |        g|        3|
# |        c|        6|
# |        c|        2|
# |        d|        3|
# |        a|        5|
# |        b|        7|
# |        c|        2|
# +---------+---------+

, но это совсем не "чище и быстрее".Лично я бы использовал RDD:

(DF1.rdd
    .flatMap(lambda row: ((val, var) for val, var in zip(row.vals, row.vars)))
    .toDF(["val", "var"])
    .show())

# +---+---+
# |val|var|
# +---+---+
# |  1|  a|
# |  2|  b|
# |  3|  c|
# |  4|  d|
# |  5|  e|
# |  6|  f|
# |  7|  g|
# |  4|  a|
# |  5|  b|
# |  3|  g|
# |  6|  c|
# |  2|  c|
# |  3|  d|
# |  5|  a|
# |  7|  b|
# |  2|  c|
# +---+---+

, но udf также будет работать.