spark scala 每个数据集输出为单行数据框答案

【问题标题】：spark scala each datasets output as a single row of dataframespark scala 每个数据集输出为单行数据框
【发布时间】：2018-06-27 17:11:26
【问题描述】：

我在一个目录中有多个 .nt (NTriples) 文件。我想读取每个数据集并将其各自的输出值存储在单行数据框中。

假设我有 dataset1.nt、dataset2.nt、...、datasetn.nt。使用以下代码读取每个数据集时：

val input = "src/main/resources/dataset1.nt"
val triplesRDD = NTripleReader.load(spark, JavaURI.create(input))
//NTripleReader reads .nt file and separates each line of dataset into subject, predicate and object     
/* My code to output number of distinct subjects, predicates and blank subjects in a dataset */

假设 dataset1 给出以下输出：

不同主题的数量：xxxx
不同谓词的数量：yy
空白科目数：zzz

假设 dataset2 给出以下输出：

不同主题的数量：aaaa
不同谓词的数量：b
空白科目数：cc

等等……

当我使用以下代码读取目录中的所有文件时：

val input = "src/main/resources/*"
val triplesRDD = NTripleReader.load(spark, JavaURI.create(input))

它给了我以下输出：

不同主题的数量：xxxx+aaaaa+... // 添加每个数据集的所有单个值
不同谓词的数量：yy+b+...
空白科目数：zzz+cc+...

但是，我希望我的输出是这样的：

Distinct Subjects | Distinct Predicates | Blank Subjects
xxxx              | yy                  | zzz
aaaaa             | b                   | cc    
...               | ...                 | ...

请告诉我如何才能实现我想要的输出。

提前致谢。

【问题讨论】：

triplesRDD 的类型是什么？这看起来像是您从 SANSA 项目中复制了代码？还是您通过 Maven 使用 SANSA RDF API？
triplesRDD 类型是 RDD[Triple]。我正在通过 Maven 使用 SANSA RDF API。我没有复制我只是使用 NTripleReader 的代码，因为我目前正在处理 SANSA STACK。

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

【解决方案1】：

我正在回答我的问题。我希望这可能对其他人有所帮助

import java.io.File
//import other necessary packages


object abc {
  var df1: DataFrame = _
  var df2: DataFrame = _         
  var df3: DataFrame = _

  def main(args: Array[String]):Unit = 
  {
    //initializing the spark session locally
    val spark = SparkSession.builder
          .master("local[*]")
          .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          .appName("abc")
          .getOrCreate()

//    creates a list of all files in a directory:
    def getListOfFiles(dir: String):List[File] = 
    {
      val path = new File("path/to/directory/")
      if (path.exists && path.isDirectory) 
      {
        path.listFiles.filter(_.isFile).toList
      } 
      else 
      {
        List[File]()
      }
    }

      val files = getListOfFiles("path/to/directory/")
      val input = ""
      for (input <- files)
      {  
      //  println(input)
        val triplesRDD = NTripleReader.load(spark, JavaURI.create(input.toString()))

        /*code to generate dataframe columns value*/

        import spark.implicits._

        if(input == files(0))
        {
            df3 = Seq(
            (column1_value, column2_value, column3_value, column4_value, column5_value, column6_value)
            ).toDF("column1_name", "column2_name", "column3_name", "column4_name", "column5_name", "column6_name")
        } 
        else
        {    
            df1 = Seq(
            (column1_value, column2_value, column3_value, column4_value, column5_value, column6_value)
            ).toDF("column1_name", "column2_name", "column3_name", "column4_name", "column5_name", "column6_name")  
            df2 = df3.union(df1)
            df3 = df2
        }
      }
      df3.show()
// import dataframe to .csv file
          df3.coalesce(1).write
          .option("header", "true")
          .csv("path/to/directory/sample.csv")
          spark.stop
      }
    }

【讨论】：