Как преобразовать, используя разнесение с пользовательской модификацией для других отдельных столбцов и в то же время - PullRequest
Новая
       18

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

0 голосов
/ 14 июня 2019

У меня есть набор данных, подобный следующему: 

Input Dataset

Id, Parent_id, Data
-----------------------
1, NULL, favorite: 3
2, NULL, favorite: 4

Output Dataset

Id, Parent_Id, Data
------------------------
1, NULL, favorite: 3
1_t1, 1, favorite: 3
1_t2, 1, favorite: 3
1_t3, 1, favorite: 3
2, NULL, favorite: 4
2_t1, 2, favorite: 4
2_t2, 2, favorite: 4
2_t3, 2, favorite: 4
2_t4, 2, favorite: 4

Как вы можете видеть выше, я пытаюсь разбить свойство любимых счетчиков столбцов данных на их отдельные строки и используяСтолбец parent_id представляет его корневую запись.

До сих пор я пытался использовать функцию Spark SQL Explode, чтобы попытаться сделать это, но я не смог заставить ее работать.

1 Ответ

1 голос
/ 15 июня 2019

Если я правильно понимаю ваш вопрос, вы пытаетесь сгенерировать / создать новые строки из существующей строки, используя число из столбца данных, и хотите сгенерировать такое количество новых строк с новыми id и parent_idуказывая на исходную запись

, если это так, то вы можете сделать это, используя операции map и flatMap, как показано ниже: 

import org.apache.spark.sql.Row

import scala.collection.mutable.ArrayBuffer

import sparkSession.sqlContext.implicits._

val input = Seq(("1", "NULL", "favorite:3"), ("2", "NULL", "favorite:4")).toDF("id", "parent_id", "data")

input.printSchema()
input.show(false)

val resultRDD = input.rdd.map(row => {
  val list = new ArrayBuffer[Row]()
  list += row

  val pointer = row.getAs[String]("data").split(":")(1).toInt

  for (index <- 1 to pointer) {
    val newId = s"${row.getAs[String]("id")}_t$index"
    list += Row.fromSeq(Seq(newId, row.getAs[String]("id"), row.getAs[String]("data")))
  }

  list
}).flatMap(_.toIterator)


val resultDF = sparkSession.createDataFrame(resultRDD, input.schema)
resultDF.show(false)

, и результат будет: 

root
 |-- id: string (nullable = true)
 |-- parent_id: string (nullable = true)
 |-- data: string (nullable = true)

+---+---------+----------+
|id |parent_id|data      |
+---+---------+----------+
|1  |NULL     |favorite:3|
|2  |NULL     |favorite:4|
+---+---------+----------+

+----+---------+----------+
|id  |parent_id|data      |
+----+---------+----------+
|1   |NULL     |favorite:3|
|1_t1|1        |favorite:3|
|1_t2|1        |favorite:3|
|1_t3|1        |favorite:3|
|2   |NULL     |favorite:4|
|2_t1|2        |favorite:4|
|2_t2|2        |favorite:4|
|2_t3|2        |favorite:4|
|2_t4|2        |favorite:4|
+----+---------+----------+
...