在scala spark中为两个不同的数据框创建视图

Posted 2023-04-15

【中文标题】在scala spark中为两个不同的数据框创建视图

【英文标题】：create view for two different dataframe in scala spark

【发布时间】：2020-09-03 09:57:31

【问题描述】：

我有一个代码 sn-p，它将读取文件路径的 Json 数组，然后合并输出并给我两个不同的表。所以我想为这两个表创建两个不同的 createOrReplaceview(name) 并且名称将在 json 数组中可用，如下所示：

      
        "source": [
            
                "name": "testPersons",
                "data": [
                "E:\\dataset\\2020-05-01\\",
                "E:\\dataset\\2020-05-02\\"
                ],
                "type": "json"
            ,
            
                "name": "testPets",
                "data": [
                "E:\\dataset\\2020-05-01\\078\\",
                "E:\\dataset\\2020-05-02\\078\\"
                ],
                "type": "json"
            
        ]
    

我的输出：

testPersons
        +---+------+
        |name  |age|
        +---+------+
        |John  |24 |
        |Cammy |20 |
        |Britto|30 |
        |George|23 |
        |Mikle |15 |
        +---+------+
 testPets
        +---+------+
        |name  |age|
        +---+------+
        |piku  |2  |
        |jimmy |3  |
        |rapido|1  |
        +---+------+

上面是我的输出和 Json 数组我的代码遍历每个数组并读取数据部分并读取数据。 但是如何更改我的以下代码为每个输出表创建一个临时视图。 例如我想创建.createOrReplaceTempView(testPersons) 和.createOrReplaceTempView(testPets) 根据 Json 数组查看名称

if (dataArr(counter)("type").value.toString() == "json") 
          val name = dataArr(counter)("name").value.toString()
          val dataPath = dataArr(counter)("data").arr
          val input = dataPath.map(item => 
            val rdd = spark.sparkContext.wholeTextFiles(item.str).map(i => "[" + i._2.replaceAll("\\.*\n0,.*\\", ",") + "]")
            spark
              .read
              .schema(Schema.getSchema(name))
              .option("multiLine", true)
              .json(rdd)
          )
          val emptyDF = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], Schema.getSchema(name))
          val finalDF = input.foldLeft(emptyDF)((x, y) => x.union(y))
          finalDF.show()

预期输出：

 spark.sql("SELECT * FROM testPersons").show()
 spark.sql("SELECT * FROM testPets").show()

它应该给我一张桌子。

【参考方案1】：

由于您已经整理好数据并将行放在DataFrames 中，并且只是想将它们作为临时视图访问，我想您正在寻找函数：

createOrReplaceGlobalTempView createOrReplaceTempView

它们可以从 DataFrame/Dataset 中调用。

df.createOrReplaceGlobalTempView("testPersons")
spark.sql("SELECT * FROM global_temp.testPersons").show()

df.createOrReplaceTempView("testPersons")
spark.sql("SELECT * FROM testPersons").show()

关于两者区别的解释，可以看一下这个question。

---

如果您尝试动态读取 JSON，请将 data 中的文件放入 DataFrames，然后将它们保存到自己的表中。

import net.liftweb.json._
import net.liftweb.json.DefaultFormats

case class Source(name: String, data: List[String], `type`: String)

val file = scala.io.Source.fromFile("path/to/your/file").mkString
implicit val formats: DefaultFormats.type = DefaultFormats
val json = parse(file)
val sourceList = (json \ "source").children
for (source <- sourceList) 
  val s = source.extract[Source]
  val df = s.data.map(d => spark.read(d)).reduce(_ union _)
  df.createOrReplaceTempView(s.name)

【讨论】：

我想从我的 Json 数组中获取视图名称

那么您是否有一个包含人和宠物（以及其他组）的 DataFrame，并且您想从中创建多个表？我没听错吗？

对于我想要一张桌子的人和我想要另一张桌子的宠物