如何在 writeStream 中访问数组类型中的元素？

Posted 2023-04-15

【中文标题】如何在 writeStream 中访问数组类型中的元素？

【英文标题】：How to access elements in a ArrayType in a writeStream?

【发布时间】：2018-03-20 04:01:53

【问题描述】：

我正在构建一个模式来接受一些数据流。它有一个带有一些元素的 ArrayType。这是我的带有 ArrayType 的 StructType：

val innerBody = StructType(
    StructField("value", LongType, false) ::
    StructField("spent", BooleanType, false) ::
    StructField("tx_index", LongType, false) :: Nil)

val prev_out = StructType(StructField("prev_out", innerBody, false) :: Nil)

val body = StructType(
StructField("inputs", ArrayType(prev_out, false), false) :: 
StructField("out", ArrayType(innerBody, false), false) :: Nill)

val schema = StructType(StructField("x",  body, false) :: Nil)

这会构建一个类似“的架构”

root
 |-- bit: struct (nullable = true)
 |    |-- x: struct (nullable = false)
 |    |    |-- inputs: array (nullable = false)
 |    |    |    |-- element: struct (containsNull = false)
 |    |    |    |    |-- prev_out: struct (nullable = false)
 |    |    |    |    |    |-- value: long (nullable = false)
 |    |    |    |    |    |-- spent: boolean (nullable = false)
 |    |    |    |    |    |-- tx_index: long (nullable = false)
 |    |    |-- out: array (nullable = false)
 |    |    |    |-- element: struct (containsNull = false)
 |    |    |    |    |-- value: long (nullable = false)
 |    |    |    |    |-- spent: boolean (nullable = false)
 |    |    |    |    |-- tx_index: long (nullable = false)

我正在尝试从模式中的“值元素”中选择值，因为它正在流入。我正在使用 writeStream 接收器。

val parsed = df.select("bit.x.inputs.element.prev_out.value")
.writeStream.format("console").start() 

我有这个，但上面的代码，但给出了错误。

消息：无法解析给定的“bit.x.inputs.element.prev_out.value” 输入列：[键、值、时间戳、分区、偏移量、 时间戳类型，主题];;

如何访问此架构中的“值”元素？

【问题讨论】：

您必须分解数组bit.x.inputs，然后选择exploded.element.prev_out.value。就这些

你能给我举个例子，我可以如何爆炸数组吗？

Rumesh 已经在下面给出了我的想法作为答案:)

***.com/questions/39739072/…

对不起，我没有看到这个例子。

【参考方案1】：

如果你有这样的数据框，先分解，然后选择会帮助你。

df.printSchema()
//root
//|-- bit: struct (nullable = true)
//|    |-- x: struct (nullable = true)
//|    |    |-- inputs: array (nullable = true)
//|    |    |    |-- element: struct (containsNull = true)
//|    |    |    |    |-- prev_out: struct (nullable = true)
//|    |    |    |    |    |-- spent: boolean (nullable = true)
//|    |    |    |    |    |-- tx_infex: long (nullable = true)
//|    |    |    |    |    |-- value: long (nullable = true)

import org.apache.spark.sql.functions._
val intermediateDf: DataFrame = df.select(explode(col("bit.x.inputs")).as("interCol"))
intermediateDf.printSchema()

//root
//|-- interCol: struct (nullable = true)
//|    |-- prev_out: struct (nullable = true)
//|    |    |-- spent: boolean (nullable = true)
//|    |    |-- tx_infex: long (nullable = true)
//|    |    |-- value: long (nullable = true)

val finalDf: DataFrame = intermediateDf.select(col("interCol.prev_out.value").as("value"))
finalDf.printSchema()
//root
//|-- value: long (nullable = true)


finalDf.show()
//+-----------+
//|      value|
//+-----------+
//|12347628746|
//|12347628746|
//+-----------+