如何使用 python 在 spark SQL 中传递变量？答案

【问题标题】：How to pass variables in spark SQL, using python?如何使用 python 在 spark SQL 中传递变量？
【发布时间】：2017-11-18 19:55:03
【问题描述】：

我正在用 python 编写 spark 代码。如何在 spark.sql 查询中传递变量？

    q25 = 500
    Q1 = spark.sql("SELECT col1 from table where col2>500 limit $q25 , 1")

目前上面的代码不起作用？我们如何传递变量？

我也试过了，

    Q1 = spark.sql("SELECT col1 from table where col2>500 limit q25='{}' , 1".format(q25))

【问题讨论】：

你试过了吗？ Q1 = spark.sql("SELECT col1 from table where col2>500 limit q25='%s' , 1"%(q25)
Does Python has a similar variable interpolation like "string #{var}" in Ruby?的可能重复
只是一个字符串...你的问题与 Spark 无关

标签： python apache-spark pyspark apache-spark-sql

【解决方案1】：

您需要在字符串格式中删除单引号和q25，如下所示：

Q1 = spark.sql("SELECT col1 from table where col2>500 limit {}, 1".format(q25))

更新：

根据您的新查询：

spark.sql("SELECT col1 from table where col2>500 order by col1 desc limit {}, 1".format(q25))

注意，SparkSQL 不支持 OFFSET，所以查询不能工作。

如果您需要添加多个变量，您可以尝试这种方式：

q25 = 500
var2 = 50
Q1 = spark.sql("SELECT col1 from table where col2>{0} limit {1}".format(var2,q25))

【讨论】：

这仍然给我不匹配的输入异常：spark.sql(SELECT col1 from table where col2>500 order by col1 desc limit {}, 1".format(q25))
',' 的输入不匹配
在SELECT前，需要双引号。
我使用过“看起来查询不适用于 ,1，即在 spark sql 中不支持 OFFSET。有什么解决方法吗？
@Viv 是的，你是对的，SparkSQL 不支持 OFFSET。你可以参考这个答案stackoverflow.com/questions/42560815/…

【解决方案2】：

如果您经常做这种事情或想让您的代码更易于重用，另一个选择是使用配置变量映射和格式选项：

configs = {"q25":10,
           "TABLE_NAME":"my_table",
           "SCHEMA":"my_schema"}
Q1 = spark.sql("""SELECT col1 from {SCHEMA}.{TABLE_NAME} 
                  where col2>500 
                  limit {q25}
               """.format(**configs))

【讨论】：

这看起来很棒。但是，它不起作用，因为它说 name: 'configs' is not defined。我错过了什么？ @大卫马多克斯
发现错误，如果你使用的是jupyter notebook，你也必须在同一个单元格中写字典

【解决方案3】：

您需要做的就是将 s（字符串插值器）添加到字符串中。这允许在字符串中直接使用变量。

val q25 = 10
Q1 = spark.sql(s"SELECT col1 from table where col2>500 limit $q25)

【讨论】：

您提供的解决方案是针对 Python 还是其他语言？好像有点不合时宜……
这似乎是 scala 实现。

【解决方案4】：

一个非常简单的解决方案是将查询存储为字符串（使用通常的 python 格式），然后将其传递给spark.sql() 函数：

q25 = 500

query = "SELECT col1 from table where col2>500 limit {}".format(q25)

Q1 = spark.sql(query)

【讨论】：