Scala:如何在循环外循环块中使用变量

Question

如何使用所有JSON文件创建数据帧，在阅读每个文件后，我需要将文件名添加为dataframe中的字段？对于循环似乎没有识别出外部循环的变量。如何克服这个问题？

for (jsonfilenames <- fileArray) {
      var df = hivecontext.read.json(jsonfilename)
      var tblLanding = df.withColumn("source_file_name", lit(jsonfilename))
    }
   // trying to create temp table from dataframe created in loop
tblLanding.registerTempTable("LandingTable") // ERROR here, can't resolved tblLanding

预先感谢
Hossain

Answer 1

我认为您是对自己编程的新手。无论如何，你去。

基本上，您在循环之前指定类型并初始化。

var df:DataFrame = null
for (jsonfilename <- fileArray) {
      df = hivecontext.read.json(jsonfilename)
      var tblLanding = df.withColumn("source_file_name", lit(jsonfilename))
    }
df.registerTempTable("LandingTable") // Getting ERROR here

update

好吧，您是编程的新手，甚至是循环。

假设filearray的值为[1.JSON，2.JSON，3.JSON，4.JSON]

因此，循环实际上通过读取4个JSON文件创建了4个数据框。您要注册为临时表的哪一个。

如果全部，

var df:DataFrame = null
var count = 0
for (jsonfilename <- fileArray) {
      df = hivecontext.read.json(jsonfilename)
      var tblLanding = df.withColumn("source_file_name", lit(jsonfilename))
      df.registerTempTable(s"LandingTable_$count")
      count++;
    }

以及此更新之前DF为空的原因，您的FileRaray为空或Spark无法读取该文件。打印并检查。

查询任何注册的LandingTable

val df2 = hiveContext.sql("SELECT * FROM LandingTable_0")

更新问题已更改为从所有JSON文件中制作单个数据帧。

var dataFrame:DataFrame = null
for (jsonfilename <- fileArray) {
   val eachDataFrame = hivecontext.read.json(jsonfilename)
   if(dataFrame == null)
      dataFrame = eachDataFrame
   else
      dataFrame = eachDataFrame.unionAll(dataFrame)
}
dataFrame.registerTempTable("LandingTable")

确保，FileRray不是空的，并且所有JSON文件都具有相同的模式。

Answer 2

// Create list of dataframes with source-file-names
val dfList = fileArray.map{ filename =>
  hivecontext.read.json(filename)
             .withColumn("source_file_name", lit(filename))
}
// union the dataframes (assuming all are same schema)
val df = dfList.reduce(_ unionAll _)  // or use union if spark 2.x
// register as table
df.registerTempTable("LandingTable")