数据帧到RDD[（字符串，字符串）]的转换

Question

我想转换一个组织。阿帕奇。火花sql。数据框到组织。阿帕奇。火花rdd。RDD[（字符串，字符串）]在数据块中。有人能帮忙吗？

背景（也欢迎使用更好的解决方案）：我有一个Kafka流，它（经过一些步骤）变成了2列数据帧。我想将其放入Redis缓存，第一列作为键，第二列作为值。

更具体地说，输入的类型是：lastContacts:org。阿帕奇。火花sql。数据帧=[序列号：字符串，上次修改：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），但都没有任何帮助。

Answer 1

您可以使用df.map（row=

例如：

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)]

请参考https://community.hortonworks.com/questions/106500/error-in-spark-streaming-kafka-integration-structu.html关于AnalysisException：具有流源的查询必须使用writeStream执行。开始（）

您需要使用query等待查询的终止。waitTermination（）以防止进程在查询处于活动状态时退出。