Как отформатировать RDD с помощью регулярного выражения, а затем сохранить его в MongoDB в Spark - PullRequest
Новая
       9

Как отформатировать RDD с помощью регулярного выражения, а затем сохранить его в MongoDB в Spark

0 голосов
/ 06 октября 2018

Исходные данные - это просто необработанный веб-журнал, агрегированный с использованием Flume и опубликованный с использованием Kafka.Например: 

60.175.130.12 - - [21/Apr/2018:20:46:35 +0800] "GET /wp-admin/edit.php HTTP/1.1" 200 13347 "http://.....php" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/11.1 Safari/605.1.15"

Я хочу использовать Spark Streaming для получения пакета журналов, а затем разделить его с помощью регулярного выражения, подобного этому: 

val regex = """^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$""".r

Разделить на форму, удобную для базы данных.: 

case class log(
            host: String,
            rfc931: String,
            username: String,
            data_time: String,
            req_method: String,
            req_url: String,
            req_protocol: String,
            statuscode: String,
            bytes: Int,
            referrer: String,
            user_agent: String)

Затем просто добавьте пакет в mongoDB.

Но у меня возникли проблемы с разбиением пакета: 

  val lines = stream.flatMap{ batch =>
  batch.value().split("\n")
}
val records = lines.map { record =>
  val regex = """^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$""".r
  val matched = regex.findAllIn(record)
  log(matched.group(1), matched.group(2), matched.group(3), matched.group(4), matched.group(5), matched.group(6), matched.group(7), matched.group(8), matched.group(9).toInt, matched.group(10), matched.group(11))
}
records.foreachRDD{ record =>
  import db.implicits._
  val record_DF = record.toDF()
  record_DF.write.mode("append").mongo()
}

Вот как я думал, что это должно бытьреализованы.Сначала разбейте поток на строки, затем сопоставьте каждую строку с помощью регулярного выражения, разделите на формат журнала и, наконец, запишите его в БД.

Сбой программы из-за 'Постоянного отсутствия совпадений' или подобных проблем с регулярным выражениемошибка при совпадении.

...

Только начинающий, нужна помощь ..

1 Ответ

0 голосов
/ 08 октября 2018

Проблема решена после изменения: 

 val records = lines.map { record =>
  val PATTERN = """^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$""".r
  val options = PATTERN.findFirstMatchIn(record)
  val matched = options.get
  log(matched.group(1), matched.group(2), matched.group(3), matched.group(4), matched.group(5), matched.group(6), matched.group(7), matched.group(8), matched.group(9).toInt, matched.group(10), matched.group(11))
}
records.print()
...