我使用spark 1.6和kafka 0.8.2.1。我正在尝试使用spark streaming从kafka获取一些数据，并对这些数据进行一些操作。为此，我应该知道所获取数据的模式，是否有某种方法可以做到这一点，或者我们可以通过使用字段名从流中获取值？
k97glaaz1#
热释光；博士，这是不可能直接（特别是与旧Spark1.6），但也不是不可能的。Kafka看到了字节，这就是spark流媒体所期望的。您必须以某种方式在固定字段上传递一些额外的信息，以获取模式（可能作为json编码的字符串）并解码另一个字段。这不是现成的，但肯定是可行的。作为建议，我会发一条信息 value 字段始终是两个字段的数据结构，模式（值字段）和值本身（json格式）。然后您可以使用其中一个from\u json函数：from\ json（e:column，schema:structtype）：column将包含json字符串的列解析为具有指定模式的structtype。鉴于 from_json 是在spark 2.1.0中添加的，您必须注册自己的自定义用户定义函数（udf），该函数将字符串值反序列化到相应的结构中（请参见 from_json 做它和复制它）。请注意 DataType 对象附带的fromjson方法可以将json编码的字符串“Map”到 DataType 可以描述你的模式。fromjson（json:string）：数据类型
value
from_json
DataType
1条答案
按热度按时间k97glaaz1#
热释光；博士，这是不可能直接（特别是与旧Spark1.6），但也不是不可能的。
Kafka看到了字节，这就是spark流媒体所期望的。您必须以某种方式在固定字段上传递一些额外的信息，以获取模式（可能作为json编码的字符串）并解码另一个字段。这不是现成的，但肯定是可行的。
作为建议，我会发一条信息
value字段始终是两个字段的数据结构，模式（值字段）和值本身（json格式）。然后您可以使用其中一个from\u json函数：
from\ json（e:column，schema:structtype）：column将包含json字符串的列解析为具有指定模式的structtype。
鉴于
from_json是在spark 2.1.0中添加的，您必须注册自己的自定义用户定义函数（udf），该函数将字符串值反序列化到相应的结构中（请参见from_json做它和复制它）。请注意
DataType对象附带的fromjson方法可以将json编码的字符串“Map”到DataType可以描述你的模式。fromjson（json:string）：数据类型