创建具有相同名称但不同数据路径的多个临时表

Posted 2021-04-19

我需要在Spark中创建一个临时表，其名称与不同的输入源相同。以下是一个例子。

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._

case class Person(name: String, age: Int)

val hr_dataframe =
  sc.textFile("/user/scvappsit/HR.txt").
     map(_.split(",")).
     map(p => Person(p(0), p(1).trim.toInt)).
     toDF()

hr_dataframe.registerTempTable("employee") //<----- employee registered

val manager_dataframe = 
  sc.textFile("/user/scvappsit/Manager.txt").
     map(_.split(",")).
     map(p => Person(p(0), p(1).trim.toInt)).
     toDF()

manager_dataframe.registerTempTable("employee") //<----- employee registered

val hr_data = sqlContext.sql("SELECT * FROM employee")

现在在上面的情况下，如果我做一个SELECT * FROM employee它将返回经理的数据而不是HR。

我需要传递临时表的别名，以便它不会覆盖HR数据。以下是一个例子。

hr_dataframe.registerTempTable("employee".alias("hr"))
sqlContext.sql("select * from hr")

答案

你可以使用隐含。如果你添加

implicit def runAlias(string: String):Alias = new Alias
class Alias {def alias(a:String):String = a }

您可以使用

hr_dataframe.registerTempTable("employee".alias("hr"))
sqlContext.sql("select * from hr")

你会看到Manager.txt的内容。员工表将保持不变，因此该解决方案与直接使用没有区别

hr_dataframe.registerTempTable("hr")

如果我们第二个隐式，我们可以为registerTempTable添加一些逻辑：

case class Person(name: String, age: Int)

import spark.implicits._
val sqlContext = spark.sqlContext

class Alias(var original: String, var al:String = "") {
  def alias(a:String):Alias =  {
    this.al = a
    this
  }
}

implicit def runAlias(string: String):Alias = new Alias(string)
implicit class DfOp(val df: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]) {
  def registerTempTable(a: Alias): org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = {
    if( sqlContext.tableNames().contains(a.original) ) {
      sqlContext.table(a.original).union(df).registerTempTable(a.original)
    }
    else {
      df.registerTempTable(a.original)
    }
    df.registerTempTable(a.al)
    df
  }
}

val hr_dataframe = sc.textFile("/tmp/HR.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
hr_dataframe.registerTempTable("employee".alias("hr"))

val manager_dataframe = sc.textFile("/tmp/Manager.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
manager_dataframe.registerTempTable("employee".alias("mgr"))

sqlContext.sql("select * from employee").show() // prints the contents of both files
sqlContext.sql("select * from hr").show() // prints the content of HR.txt
sqlContext.sql("select * from mgr").show() // prints the context of Manager.txt

第一个隐式只记录原始表名和别名。

然后第二个隐式将采用原始表并检查是否已存在具有此名称的表。如果是这样，它将合并当前正在注册的表与现有表，否则它将只创建一个新表。之后，将使用别名创建第二个表。

在将两个表（HR和Manager）及其各自的别名注册后，将有三个表：

• 包含两个文件中的行的employee
• hr包含HR.txt中的行
• mgr，包含Manager.txt中的行

另一答案

没有为表名提供别名的功能。在我看来，你有3个选择，那是因为我不理解你的要求。因此，基于您的潜在愿望，如果：

1. 您希望将所有数据合并到一个表中：在仍处于RDD中时将两个RDD合并，然后将结果注册为临时表 hr_dataframe.union（manager_dataframe）.registerTempTable（ “雇员”）
2. 您希望它是2个单独的表：然后您需要调用一个经理，而另一个员工
3. 您希望它们都被称为员工但仍然是分开的：然后您将需要使用不同的模式名称并在“hr.employee”下保存一个表，在“manager.employee”下保存另一个表。您可能需要预先创建这些模式名称。