apache-spark - Spark sql如何为输入DataFrame中的每条记录循环执行sql命令

Question

Spark sql如何为输入DataFrame中的每条记录循环执行sql命令

我有一个具有以下架构的 DataFrame

     %> input.printSchema
        root
        |-- _c0: string (nullable = true)
        |-- id: string (nullable = true)

我有另一个 DataFrame，我需要在其上执行 sql 命令

        val testtable = testDf.registerTempTable("mytable")

        %>testDf.printSchema
            root
            |-- _1: integer (nullable = true)

        sqlContext.sql(s"SELECT * from mytable WHERE _1=$id").show()

$id 应该来自输入 DataFrame 并且应该对所有输入表 id 执行 sql 命令

Answer 1

假设您可以使用一个新的 DataFrame，其中包含 testDf 中存在的所有行，这些行与 input 的 id 列中存在的值相匹配>，您可以执行内部联接操作，如 Alberto 所述:

val result = input.join(testDf, input("id") == testDf("_1"))
result.show()

现在，如果您想要为 testDf 中存在的每个不同值创建一个新的、不同的 DataFrame，那么问题就更难了。如果是这种情况，我建议您确保可以将查找表中的数据收集为本地列表，这样您就可以遍历它的值并为您已经想到的每个值创建一个新的 DataFrame(这 不推荐):

val localArray: Array[Int] = input.map { case Row(_, id: Integer) => id }.collect
val result: Array[DataFrame] = localArray.map {
  i => testDf.where(testDf("_1") === i)
}

无论如何，除非查找表非常小，否则我建议您调整您的逻辑以使用我的第一个示例中的单个连接 DataFrame。