Я тестирую искру, используя scala. Обычно мы читаем json файлы, которые нужно манипулировать, как в следующем примере:
test.json:
{"a":1,"b":[2,3]}
val test = sqlContext.read.json("test.json")
Как преобразовать его в следующий формат:
{"a":1,"b":2}
{"a":1,"b":3}
Вы можете использовать функцию explode:
scala> import org.apache.spark.sql.functions.explode
import org.apache.spark.sql.functions.explode
scala> val test = sqlContext.read.json(sc.parallelize(Seq("""{"a":1,"b":[2,3]}""")))
test: org.apache.spark.sql.DataFrame = [a: bigint, b: array<bigint>]
scala> test.printSchema
root
|-- a: long (nullable = true)
|-- b: array (nullable = true)
| |-- element: long (containsNull = true)
scala> val flattened = test.withColumn("b", explode($"b"))
flattened: org.apache.spark.sql.DataFrame = [a: bigint, b: bigint]
scala> flattened.printSchema
root
|-- a: long (nullable = true)
|-- b: long (nullable = true)
scala> flattened.show
+---+---+
| a| b|
+---+---+
| 1| 2|
| 1| 3|
+---+---+
часто предлагается использовать explode, но из нетипизированного API DataFrame и с учетом того, что вы используете набор данных, я думаю, что лучше подойдет оператор flatMap (см. org.apache.spark.sql.Dataset).
flatMap[U](func: (T) ⇒ TraversableOnce[U])(implicit arg0: Encoder[U]): Dataset[U]
(Специфично для Scala) Возвращает новый набор данных, сначала применяя функцию ко всем элементам этого набора данных, а затем выравнивая результаты.
Вы можете использовать его следующим образом:
val ds = Seq(
(0, "Lorem ipsum dolor", 1.0, Array("prp1", "prp2", "prp3")))
.toDF("id", "text", "value", "properties")
.as[(Integer, String, Double, scala.List[String])]
scala> ds.flatMap { t =>
t._4.map { prp =>
(t._1, t._2, t._3, prp) }}.show
+---+-----------------+---+----+
| _1| _2| _3| _4|
+---+-----------------+---+----+
| 0|Lorem ipsum dolor|1.0|prp1|
| 0|Lorem ipsum dolor|1.0|prp2|
| 0|Lorem ipsum dolor|1.0|prp3|
+---+-----------------+---+----+
// or just using for-comprehension
for {
t <- ds
prp <- t._4
} yield (t._1, t._2, t._3, prp)