How to convert spark log files into one CSV File

Question

I have a collection of spark Apps Log files, I want Each file App Name, Submission Time, Completion Time, and Accumulables metrics to be added as a single line in one CSV file. using SPARK/SCALA Edit: I am sorry but one Spark application log file is so huge to put in here and also so complicated some metrics are updated repeatedly for each job and I need the total of all of them -the final ones not the updated- here what I tried until now

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()
  }
}

I am not finished yet but getting better results with more modifications

Welcome to stackoverflow! Your question is not well defined, please expand the problem statement and show a short but complete example of what have you tried so far. Check this for reference: http://jonskeet.uk/csharp/complete.html — vsnyc, Jul 04 '18 at 19:53

score 0 · Answer 1 · answered Jul 05 '18 at 00:01

I understood your question a little and on the basis of my understanding I am answering. I hope you can structure your question further and I might answer your question in detail.

//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")

Hi Pardeep, your `file_name.csv` is not a [standard CSV](https://stackoverflow.com/q/58142220/287948)... I have similar problem at [this question](https://stackoverflow.com/q/59079862/287948), and the solution is to use Log4j or PrintWriter. Perhaps you can express here a solution using Log4j. — Peter Krauss, Nov 30 '19 at 12:06

How to convert spark log files into one CSV File

1 Answers1