Разобрать неструктурированные блоги в структурированный формат

Question

У меня есть файл, который содержит такие записи, как,

Ввод (необработанная запись журнала):

50.57.190.149 - - [22 / апр / 2012: 07: 12: 41 +0530] "GET /computers/laptops.html?brand=819 HTTP / 1.0" 200 12530 "-" "-"

Выход (обработанная запись журнала):

50.57.190.149 - - 22 / апр / 2012: 07: 12: 41 +0530 GET /computers/laptops.html?brand=819 компьютеры HTTP / 1.0 - - laptops.html марка = 819 200 12530 - -

Формат входных данных:

1. удаленный IP
2. удаленный лог-имя
3. Пользователь
4. время
5. запрос строка
6. статус-код
7. байт-строка
8. агент пользователя
9. направление

вот мой код:

object unStructuredToStructured {

  def main(args : Array[String]){ 

  val spark = SparkSession.builder().appName("unStructuredToStructured").master("local[*]").getOrCreate()
 import spark.implicits._

 val rdd1 = spark.read.textFile("C:\\Users\\LENOVO\\Downloads\\Veeresh_study\\DataSet_from_OldSessions\\weblogs\\weblogs_1_rec.txt").rdd

 val schemaString = "remote_IP remote_log_name user time request_string status_code byte_string user_agent referral"

 val fields = schemaString.split(" ")
   .map(fieldName => StructField(fieldName, StringType, nullable = true))

 val schema = StructType(fields)


 val rowRDD = rdd1.map(x => x.split(" "))
   .map(attributes => Row(attributes(0), attributes(1), attributes(2), attributes(3), attributes(4), attributes(5), attributes(6), attributes(7), attributes(8)))

 val data = spark.createDataFrame(rowRDD, schema)
 data.show()

}
}

Выход:

Вот вывод, который я получаю

Как видно из изображения,

мы используем пробел в качестве удалителя и значение одного поля разбивается на несколько столбцов (поскольку значение-поле содержит пробел в нем)

ex: В идеале столбец " time " должен иметь значение " [22 / Apr / 2012: 07: 12: 41 +0530] ", но здесь он разбивается на два столбца, например " time " и " request_string "

аналогично значение request_string ("GET /computers/laptops.html?brand=819 HTTP / 1.0") разделяется на " status_code ", " byte_string"и " user_agent"

помогите разобрать значения полей, игнорируя пробел внутри значения поля

Answer 1

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

object unStructuredToStructured {
  def main(args : Array[String]){

    val spark = SparkSession.builder().appName("unStructuredToStructured").master("local[*]").getOrCreate()
   import spark.implicits._

   val rdd1 = spark.read.textFile("C:\\Users\\LENOVO\\Downloads\\Veeresh_study\\DataSet_from_OldSessions\\weblogs\\weblogs_10_lakh_rec.txt").rdd

    val schemaString = "remote_IP remote_log_name user time request_string status_code byte_string user_agent referral"

    val fields = schemaString.split(" ")
    .map(fieldName => StructField(fieldName, StringType, nullable = true))

    val schema = StructType(fields)


    val rowRDD = rdd1.map(x => x.split(" "))
    .map(attributes => 
      Row(attributes(0), attributes(1), attributes(2), attributes(3), attributes(4), attributes(5), attributes(6), attributes(7), attributes(8))
      )
   // rowRDD.foreach(println)

    def combiner(arr : Array[String])  = {

     val len = arr.length
     if (len <= 15)
     {

       var val0  = arr(0)
     var val1  = arr(1)
     var val2 = arr(2)
     var val3  = arr(3).concat(arr(4))
     var val4  = arr(5).concat(arr(6)).concat(arr(7))
     var val5  = arr(8)
     var val6 = arr(9)
       var last = arr.last
       var value : String = null


    for(i <- 10 until len-1) {
     if(value == null)
           value = arr(i)
      else           
      value = value.concat(arr(i))

    }

       Row(val0, val1, val2,val3,val4,val5,val6,value,last)
     }
     else 
     {

     var val0  = arr(0)
     var val1  = arr(1)
     var val2 = arr(2)
     var val3  = arr(3).concat(arr(4))
     var val4  = arr(5).concat(arr(6)).concat(arr(7))
     var val5  = arr(8)
     var val6 = arr(9)
     var val7  = arr(10).concat(arr(11)).concat(arr(12)).concat(arr(13))
     .concat(arr(14)).concat(arr(15)).concat(arr(16)).concat(arr(17)).concat(arr(17)).concat(arr(18)).concat(arr(19))
     var val8  = arr(20)
     var last = arr.last

    //val len1 = arr.length
    var value : String = null
    for(i <- 10 until len-1) {
     if(value == null)
           value = arr(i)
      else           
      value = value.concat(arr(i))

    }
  // Row(len,val0, val1, val2,val3,val4,val5,val6,val7,val8)
 Row(val0, val1, val2,val3,val4,val5,val6,value,last)
     }
      }

     val rowRDD1 = rdd1.map(x => x.split(" "))
    .map{attributes => 
     combiner(attributes)
     }
      rowRDD1.foreach(println)

    val data = spark.createDataFrame(rowRDD1, schema)
   data.show()
  }
}

и вот скриншот o / p: Вот вывод