基于spark（scala）中条件的联合数据帧

Question

我有一个包含 4 个子文件夹的文件夹，其中包含镶木地板文件 文件夹->A.parquet、B.parquet、C.parquet、D.parquet（子文件夹）。我的要求是我想根据我提供给该方法的文件名来合并数据框。 我正在用代码做

val df = listDirectoriesGetWantedFile(folderPath,sqlContext,A,B)
def listDirectoriesGetWantedFile(folderPath: String, sqlContext: SQLContext, str1: String, str2: String): DataFrame = {
    var df: DataFrame = null
    val sb = new StringBuilder
    sb.setLength(0)
    var done = false
    val path = new Path(folderPath)
    if (fileSystem.isDirectory(path)) {
        var files = fileSystem.listStatus(path)
        for (file <- files) {
        if (file.getPath.getName.contains(str) && !done) {
          sb.append(file.getPath.toString())
          sb.append(",")
          done = true
        } else if (file.getPath.getName.contains(str2)) {
          sb.append(file.getPath.toString())
        }
    }
}

但我需要拆分 sb 然后合并数据框。我无法找到解决方案。我该如何处理它并解决

Answer 1

如果我理解你的问题，你可以简单地做这样的事情：

def listDirectoriesGetWantedFile(path: String, 
                                 sqlContext: SQLContext, 
                                 folder1: String, 
                                 folder2: String): DataFrame = {
  val df1 = sqlContext.read.parquet(s"$path/$folder1")
  val df2 = sqlContext.read.parquet(s"$path/$folder2")
  df1.union(df2)
}

---

编辑

通过使用 Hadoop 文件系统，您可以检查文件夹上的路径是否存在。所以你可以尝试这样的事情：

def listDirectoriesGetWantedFile(path: String, sqlContext: SQLContext, folders: Seq[String]): DataFrame = {
    val conf = new Configuration()
    val fs = FileSystem.get(conf)

    val existingFolders = folders
      .map(folder => new Path(s"$path/$folder"))
      .filter(fs.exists(_))
      .map(_.toString)

    if (existingFolders.isEmpty) {
      sqlContext.emptyDataFrame
    } else {
      sqlContext.read.parquet(existingFolders: _*)
    }
  }