Как выбрать столбцы на основе значения столбца?

Question

Моя схема данных приведена ниже:

root
 |-- value: struct (nullable = true)
 |    |-- before: struct (nullable = true)
 |    |    |-- id: long (nullable = false)
 |    |    |-- name: string (nullable = false)
 |    |-- after: struct (nullable = true)
 |    |    |-- id: long (nullable = false)
 |    |    |-- name: string (nullable = false)
 |    |-- op: string (nullable = false)
 |    |-- ts_ms: long (nullable = true)

Я хочу, чтобы схема выбирала до или после базирования по значению столбца ("op")

Так что-то вроде

root
 |    |-- id: long (nullable = false) //from before if col("op")==='D' else for after 
 |    |-- name: string (nullable = false) //from before if col("op")==='D' else for after 
 |    |-- op: string (nullable = false)

Мой текущий код выбирает после, но мне нужно, чтобы он был условным, чтобы при удалении событий он мог выбирать после раздела.

dataFrame
      .select(from_confluent_avro(col("Value"), valueRegistryConfig) as 'value)
      .select("value.op", "value.after.*")

До и После может иметь n полей с полем то же имя. Кроме того, до и после, я также хочу, чтобы поле op.

Возможные значения поля 'op':

• c
• d
• ты

Answer 1

С ABRiS - Avro Bridge для Spark нормально работали:

import za.co.absa.abris.avro.functions.from_confluent_avro
dataFrame
  .select(from_confluent_avro(col("Value"), valueRegistryConfig) as 'value)
  .withColumn("value_after_before",
    when(col("value.op") === "d", col("value.before"))
      .otherwise(col("value.after")))
  .select("value.op", "value_after_before.*")

Answer 2

.select(when($"value.op" === "D", $"before.id").otherwise($"after.id")), et c.