我想在Databricks中将org.apache.spark.sql.DataFrame
转换为org.apache.spark.rdd.RDD[(String, String)]
。有人可以帮忙吗?
背景(也是一个更好的解决方案也是受欢迎的):我有一个Kafka流(在一些步骤之后)变成一个2列数据帧。我想将它放入Redis缓存中,第一列作为键,第二列作为值。
更具体地说,输入的类型是:lastContacts: org.apache.spark.sql.DataFrame = [serialNumber: string, lastModified: bigint]
。我尝试按如下方式放入Redis:
sc.toRedisKV(lastContacts)(redisConfig)
错误消息如下所示:
notebook:20: error: type mismatch;
found : org.apache.spark.sql.DataFrame
(which expands to) org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
required: org.apache.spark.rdd.RDD[(String, String)]
sc.toRedisKV(lastContacts)(redisConfig)
我已经玩了一些想法(比如功能.rdd
)但没有帮助。
如果要将行映射到不同的RDD元素,可以使用df.map(row => ...)将数据帧转换为RDD。
例如:
val df = Seq(("table1",432),
("table2",567),
("table3",987),
("table1",789)).
toDF("tablename", "Code").toDF()
df.show()
+---------+----+
|tablename|Code|
+---------+----+
| table1| 432|
| table2| 567|
| table3| 987|
| table1| 789|
+---------+----+
val rddDf = df.map(r => (r(0), r(1))).rdd // Type:RDD[(Any,Any)]
OR
val rdd = df.map(r => (r(0).toString, r(1).toString)).rdd //Type: RDD[(String,String)]
有关AnalysisException,请参阅https://community.hortonworks.com/questions/106500/error-in-spark-streaming-kafka-integration-structu.html:必须使用writeStream.start()执行带有流源的查询
您需要使用query.awaitTermination()等待查询终止。以防止在查询处于活动状态时退出进程。