Apache Spark SQL 标识符预期异常答案

【问题标题】：Apache Spark SQL identifier expected exceptionApache Spark SQL 标识符预期异常
【发布时间】：2015-12-03 17:45:52
【问题描述】：

我的问题与这个问题非常相似：Apache Spark SQL issue : java.lang.RuntimeException: [1.517] failure: identifier expected 但我就是不知道我的问题出在哪里。我使用 SQLite 作为数据库后端。连接和简单的选择语句工作正常。

违规行：

val df = tableData.selectExpr(tablesMap(t).toSeq:_*).map(r => myMapFunc(r))

tablesMap 包含作为键的表名和作为表达式的字符串数组。打印出来的数组是这样的：

WrappedArray([My Col A], [ColB] || [Col C] AS ColB)

表名也包含在方括号中，因为它包含空格。我得到的例外：

Exception in thread "main" java.lang.RuntimeException: [1.1] failure: identifier expected

我已经确定不使用任何 Spark Sql 关键字。在我看来，这段代码失败有两个可能的原因：1）我以某种方式错误地处理了列名中的空格。 2）我处理连接错误。

我正在使用类似于 CSV 的资源文件，其中包含我想在我的表上评估的表达式。除了这个文件，我想允许用户在运行时指定额外的表和它们各自的列表达式。该文件如下所示：

TableName,`Col A`,`ColB`,CONCAT(`ColB`, ' ', `Col C`)

显然这不起作用。不过，我想重用这个文件，当然是修改过的。我的想法是用字符串数组中的表达式映射列，就像现在一样，映射到一系列 spark 列。（这对我来说是我能想到的唯一解决方案，因为我想避免仅针对这一功能引入所有配置单元依赖项。）我将为我的表达式引入一个小语法，以使用 $ 和concat 和 as 等函数的一些关键字。但我怎么能这样做呢？我试过这样的东西，但它离编译还很远。

def columnsMapFunc( expr: String) : Column = {
    if(expr(0) == '$')
        return expr.drop(1)
    else
        return concat(extractedColumnNames).as(newName)
}

【问题讨论】：

标签： scala sqlite apache-spark apache-spark-sql

【解决方案1】：

一般来说，使用包含空格的名称会带来问题，但用反引号替换方括号应该可以解决问题：

val df = sc.parallelize(Seq((1,"A"), (2, "B"))).toDF("f o o", "b a r")
df.registerTempTable("foo bar")

df.selectExpr("`f o o`").show

// +-----+
// |f o o|
// +-----+
// |    1|
// |    2|
// +-----+

sqlContext.sql("SELECT `b a r` FROM `foo bar`").show

// +-----+
// |b a r|
// +-----+
// |    A|
// |    B|
// +-----+

对于连接，您必须使用concat 函数：

df.selectExpr("""concat(`f o o`, " ", `b a r`)""").show

// +----------------------+
// |'concat(f o o, ,b a r)|
// +----------------------+
// |                   1 A|
// |                   2 B|
// +----------------------+

但在 Spark 1.4.0 中它需要 HiveContext。

实际上，我会在加载数据后简单地重命名列

df.toDF("foo", "bar")
// org.apache.spark.sql.DataFrame = [foo: int, bar: string]

并使用函数而不是表达式字符串（concat 函数仅在 Spark >= 1.5.0 中可用，对于 1.4 及更早版本，您需要 UDF）：

import org.apache.spark.sql.functions.concat

df.select($"f o o", concat($"f o o", lit(" "), $"b a r")).show

// +----------------------+
// |'concat(f o o, ,b a r)|
// +----------------------+
// |                   1 A|
// |                   2 B|
// +----------------------+

还有concat_ws函数以分隔符作为第一个参数：

df.selectExpr("""concat_ws(" ", `f o o`, `b a r`)""")
df.select($"f o o", concat_ws(" ", $"f o o", $"b a r"))

【讨论】：