Как преобразовать файлы журнала искры в один файл CSV

Question

У меня есть коллекция файлов журнала приложений Spark, я хочу, чтобы каждый файл Имя приложения, Время отправки, Время завершения и Накопительные метрики были добавлены в виде одной строки в одном файле CSV. используя SPARK / SCALA Редактировать: Извините, но один файл журнала приложения Spark настолько огромен, чтобы его можно было вставить сюда, а также настолько сложен, что некоторые показатели неоднократно обновляются для каждой работы, и мне нужно общее количество всех из них - последние, а не обновленные - вот что я пытался до Теперь

import org.apache.log4j._
import org.apache.spark.sql._


object LogToCSV {
  val Logs= "SparkAppName, SubmissionTime, CompletionTime,ExecutorDeserializeCpuTime,ResultSize,ShuffleReadRemoteBytesRead, ShuffleReadFetchWaitTime,MemoryBytesSpilled,ShuffleReadLocalBytesRead,ExecutorDeserializeTime,PeakExecutionMemory,ExecutorCpuTime, ShuffleReadLocalBlocksFetched,JVMGCTime,ShuffleReadRemoteBytesReadToDisk,ShuffleReadRecordsRead,DiskBytesSpilled,ExecutorRunTime,ShuffleReadRemoteBlocksFetched,Result"
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    Logger.getLogger("akka").setLevel(Level.ERROR)
    val ss = SparkSession
      .builder
      .appName("SparkSQLDFjoin")
      .master("local[*]")
      .getOrCreate()
    import ss.implicits._
ScalaWriter.Writer.Write(Logs, "Results.csv")
    val Dir = ss.sparkContext.wholeTextFiles("/home/rudaini/Desktop/Thesis/Results/Results/Tesx/*")
    println(Dir.count())
    Dir.foreach(F =>{
      var SparkAppName = ""
      var SubmissionTime: Double = 0
      var CompletionTime: Double = 0
      var ExecutorDeserializeCpuTime: Double = 0
      var ResultSize = ""
      var ShuffleReadRemoteBytesRead = ""
      var ShuffleReadFetchWaitTime = ""
      var MemoryBytesSpilled = ""
      var ShuffleReadLocalBytesRead = ""
      var ExecutorDeserializeTime = ""
      var PeakExecutionMemory = ""
      var ExecutorCpuTime = ""
      var ShuffleReadLocalBlocksFetched = ""
      var JVMGCTime = ""
      var ShuffleReadRemoteBytesReadToDisk = ""
      var ShuffleReadRecordsRead = ""
      var DiskBytesSpilled = ""
      var ExecutorRunTime = ""
      var ShuffleReadRemoteBlocksFetched = ""
      var Result = ""

      F.toString().split("\n").foreach(L =>{
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}

        if(L.contains("ApplicationStart")){
          SubmissionTime = L.substring(L.indexOf("Timestamp")+11,
            L.indexOf(",\"User\":\"")).toDouble}

        if(L.contains("ApplicationEnd")){
          CompletionTime = L.substring(L.indexOf("Timestamp")+11,L.indexOf("Timestamp")+24).toDouble}

        if(L.contains("SparkSubmit.scala")){
          ExecutorDeserializeCpuTime = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")).toDouble}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}
        if(L.contains("spark.app.name")){
          SparkAppName = L.substring(L.indexOf("app.name")+11,
            L.indexOf("spark.scheduler")-3)}


      })
      val LineX  = SparkAppName +","+ SubmissionTime +","+ CompletionTime +","+ ExecutorDeserializeCpuTime +","+ ResultSize +","+ ShuffleReadRemoteBytesRead +","+ ShuffleReadFetchWaitTime +","+ MemoryBytesSpilled +","+
        ShuffleReadLocalBytesRead +","+ ExecutorDeserializeTime +","+ PeakExecutionMemory +","+ ExecutorCpuTime +","+
        ShuffleReadLocalBlocksFetched +","+ JVMGCTime +","+ ShuffleReadRemoteBytesReadToDisk +","+
        ShuffleReadRecordsRead +","+ DiskBytesSpilled +","+ ExecutorRunTime +","+ ShuffleReadRemoteBlocksFetched +","+
        Result

      ScalaWriter.Writer.Write(LineX, "Results.csv")
    })    
    ss.stop()
  }
}

Я еще не закончил, но получаю лучшие результаты с большим количеством модификаций

Answer 1

Я немного понял ваш вопрос и, исходя из своего понимания, отвечаю. Я надеюсь, что вы можете структурировать свой вопрос дальше, и я мог бы ответить на ваш вопрос подробно.

//define all dataframes globally
var df1: DataFrame = _
var df2: DataFrame = _      
var df3: DataFrame = _
// define main function
//initialize spark session
//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)
  {
    // code to extract log file data (I can help you further if you will explain your problem further)
   // load your log file data into a dataframe
 import spark.implicits._

    if(input == files(0))
    {
        df1 = Seq(
        (App Name.value, Submission Time.value, Completion Time.value, Accumulables metrics.value)
        ).toDF("App Name", "Submission Time", "Completion Time", "Accumulables metrics")
    } 
    else
    {    
        df2 = Seq(
        (App Name.value, Submission Time.value, Completion Time.value, Accumulables metrics.value)
        ).toDF("App Name", "Submission Time", "Completion Time", "Accumulables metrics")  
        df3 = trainingDF.union(df2)
        df1 = df3
    }
  }

  //  import dataframe to .csv file
  df1.coalesce(1).write
  .option("header", "true")
  .csv("path/to/directory/file_name.csv")