Переименовать ключ во вложенной схеме Spark DataFrame (Scala)

По сути, вы должны создать выражение Column, которое приведёт ваш ввод к типу с очищенными именами полей. Для этого вы можете использовать функцию org.apache.spark.sql.functions.struct, которая позволяет комбинировать другие Column для создания столбца структурного типа. Нечто подобное должно работать:

  import org.apache.spark.sql.{functions => f}

  def sanitizeName(s: String): String = s.replace(" ", "_")

  def sanitizeFieldNames(st: StructType, context: String => Column): Column = f.struct(
    st.fields.map { sf =>
      val sanitizedName = sanitizeName(sf.name)
      val sanitizedField = sf.dataType match {
        case struct: StructType =>
          val subcontext = context(sf.name)
          sanitizeFieldNames(struct, subcontext(_))
        case _ => context(sf.name)
      }
      sanitizedField.as(sanitizedName)
    }: _*
  )

Вы используете это так:

val df: DataFrame = ...

val appFieldType = df.schema("app").asInstanceOf[StructType]  // or otherwise obtain the field type
df.withColumn(
  "app",
  sanitizeFieldNames(appFieldType, df("app")(_))
)

Для вашего типа эта рекурсивная функция будет возвращать столбец типа

f.struct(
  df("app")("environment").as("environment"),
  df("app")("name").as("name"),
  f.struct(
    df("app")("type")("word tier").as("word_tier"),
    df("app")("type")("level").as("level")
  ).as("type")
)

[1120 ], который затем присваивается полю «app», заменяя то, что там присутствует.

Однако у этого решения есть ограничение. Он не поддерживает вложенные массивы или карты: если у вас есть схема со структурами внутри массивов или карт, этот метод не будет преобразовывать какие-либо структуры внутри массивов и карт. Тем не менее, в Spark 2.4 они добавили функции, которые выполняют операции с коллекциями, поэтому возможно, что в Spark 2.4 эта функция может быть обобщена для поддержки вложенных массивов и карт.

Наконец, можно делать то, что вы хотите с mapPartitions. Во-первых, вы пишете рекурсивный метод, который очищает только StructType вашего поля:

def sanitizeType(dt: DataType): DataType = dt match {
  case st: StructType => ...  // rename fields and invoke recursively
  case at: ArrayType => ...  // invoke recursively
  case mt: MapType => ...  // invoke recursively
  case _ => dt  // simple types do not have anything to sanitize
}

Во-вторых, вы применяете очищенную схему к вашему фрейму данных. Есть два основных способа сделать это: безопасный mapPartitions и один, который использует внутренний Spark API.

С mapPartitions это просто:

df.mapPartitions(identity)(RowEncoder(sanitizeType(df.schema)))

Здесь мы применяем операцию mapPartitions и явно указываем выходной кодер. Помните, что схемы в Spark не свойственны данным: они всегда связаны с конкретным кадром данных. Все данные внутри фрейма данных представлены в виде строк без меток на отдельных полях, только позиции. Пока ваша схема имеет одинаковые типы на тех же позициях (но с потенциально разными именами), она должна работать так, как вы ожидаете.

mapPartitions действительно приводит к нескольким дополнительным узлам в логическом плане. Чтобы избежать этого, можно создать экземпляр Dataset[Row] непосредственно с помощью конкретного кодировщика:

new Dataset[Row](df.sparkSession, df.queryExecution.logical, RowEncoder(sanitizeType(df.schema)))

Это позволит избежать ненужного mapPartitions (что, как правило, приводит к десериализации-map-serialize шаги в плане выполнения запроса), но это может быть небезопасно; Я лично не проверял это сейчас, но это могло сработать для вас.