创建数据框时输出不适当 [英] inappropriate output while creating a dataframe
问题描述
我正在尝试使用scala应用程序从kafka主题流式传输数据。我能够从该主题中获取数据,但是如何从中创建数据框架?
I'm trying to stream the data from kafka topic using scala application.I'm able to get the data from the topic, but how to create a data frame out of it?
这里是数据(字符串,字符串格式)
Here is the data(in string,string format)
{
"action": "AppEvent",
"tenantid": 298,
"lat": 0.0,
"lon": 0.0,
"memberid": 16390,
"event_name": "CATEGORY_CLICK",
"productUpccd": 0,
"device_type": "iPhone",
"device_os_ver": "10.1",
"item_name": "CHICKEN"
}
我尝试了几种方法,但效果不理想。
I tried few ways to do it, but it is not yielding satisfactory results.
+--------------------+ | _1|
+--------------------+ |{"action":"AppEve...| |{"action":"AppEve...| |{"action":"AppEve...| |{"action":"AppEve...| |{"action":"AppEve...|
|{"action":"AppEve...| |{"action":"AppEve...| |{"action":"AppEve...|
|{"action":"AppEve...| |{"action":"AppEve...|
任何人都可以告诉如何进行映射,以便每个字段都可以进入像表这样的单独列。
Can anyone tell How to do the mapping so that each field goes in to a seperate column like a table. the data is in avro format.
这是从主题中获取数据的代码。
here is the code which is getting the data from the topic.
val ssc = new StreamingContext(sc, Seconds(2))
val kafkaConf = Map[String, String]("metadata.broker.list" -> "####",
"zookeeper.connect" -> "########",
"group.id" -> "KafkaConsumer",
"zookeeper.connection.timeout.ms" -> "1000000")
val topicMaps = Map("fishbowl" -> 1)
val messages = KafkaUtils.createStream[String, String,DefaultDecoder, DefaultDecoder](ssc, kafkaConf, topicMaps, StorageLevel.MEMORY_ONLY_SER).map(_._2)
请指导我如何使用foreachRDD函数和map()创建适当的数据框
please guide me how to use foreachRDD func and map() to create a proper data frame
推荐答案
根据rdd创建一个数据框,而不考虑其案例类架构。
使用以下逻辑
To create a dataframe out of an rdd irrespective of its case class schema. Use this below logic
stream.foreachRDD(
rdd => {
val dataFrame = sqlContext.read.json(rdd.map(_._2))
dataFrame.show()
})
此处流是从kafkaUtils.createStream()创建的rdd。
这篇关于创建数据框时输出不适当的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!