Как преобразовать некоторые атрибуты JSON в строки, используя фреймы данных в спарк

Question

Я новичок и пытаюсь решить следующую проблему. Любая помощь высоко ценится.

У меня есть следующий Json.

{
  "index": "identity",
  "type": "identity",
  "id": "100000",
  "source": {
    "link_data": {
      "source_Id": "0011245"
    },
    "attribute_data": {
      "first": {
        "val": [
          true
        ],
        "updated_at": "2011"
      },
      "second": {
        "val": [
          true
        ],
        "updated_at": "2010"
      }
    }
  }
}

Атрибуты в "attribute_data" могут отличаться. он может иметь другой атрибут, скажем, «третий»

Я ожидаю результата в следующем формате:

_index _type _id        source_Id   attribute_data   val      updated_at
ID     ID    randomid   00000       first            true    2000-08-08T07:51:14Z
ID     ID    randomid   00000       second           true    2010-08-08T07:51:14Z

Я попробовал следующий подход.

val df = spark.read.json("sample.json")

val res =  df.select("index","id","type","source.attribute_data.first.updated_at", "source.attribute_data.first.val", "source.link_data.source_id");

Он просто добавляет новый столбец, а не строки, как показано

 index     id     type          updated_at    val    source_id
identity 100000  identity        2011        [true]   0011245

Answer 1

Здесь вы идете с решением. Не стесняйтесь спрашивать, если вам нужно что-то понять:

val data = spark.read.json("sample.json")
val readJsonDf = data.select($"index", $"type", $"id", $"source.link_data.source_id".as("source_id"), $"source.attribute_data.*")
readJsonDf.show()

Начальный выход:

+--------+--------+------+---------+--------------------+--------------------+ | index| type| id|source_id| first| second| +--------+--------+------+---------+--------------------+--------------------+ |identity|identity|100000| 0011245|[2011,WrappedArra...|[2010,WrappedArra...| +--------+--------+------+---------+--------------------+--------------------+

Затем я сделал динамическое преобразование, используя следующие строки кода:

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._

def transposeColumnstoRows(df: DataFrame, constantCols: Seq[String]): DataFrame = {
  val (cols, types) = df.dtypes.filter{ case (c, _) => !constantCols.contains(c)}.unzip

  //a check if the required columns that needs to be transformed to rows are of the same structure
  require(types.distinct.size == 1, s"${types.distinct.toString}.length != 1")

  val keyColsWIthValues = explode(array(
  cols.map(c => struct(lit(c).alias("columnKey"), col(c).alias("value"))): _*
  ))

  df.select(constantCols.map(col(_)) :+ keyColsWIthValues.alias("keyColsWIthValues"): _*)
}



val newDf = transposeColumnstoRows(readJsonDf, Seq("index","type","id","source_id"))
val requiredDf = newDf.select($"index",$"type",$"id",$"source_id",$"keyColsWIthValues.columnKey".as("attribute_data"),$"keyColsWIthValues.value.updated_at".as("updated_at"),$"keyColsWIthValues.value.val".as("val"))
requiredDf.show()

Окончательный вывод:

|   index|    type|    id|source_id|attribute_data|updated_at|   val|
+--------+--------+------+---------+--------------+----------+------+
|identity|identity|100000|  0011245|         first|      2011|[true]|
|identity|identity|100000|  0011245|        second|      2010|[true]|

Надеюсь, это решит вашу проблему!

Answer 2

Попробуйте следующее:

import org.apache.spark.sql.functions._

import spark.implicits._
val df = spark.read.json("sample.json")

df.select($"id", $"index", $"source.link_data.source_Id".as("source_Id"),$"source.attribute_data.first.val".as("first"), explode($"source.attribute_data.second.val").as("second"), $"type")
.select($"id", $"index", $"source_Id", $"second", explode($"first"), $"type").show