streaming

val lines = kafkaStream.flatMap { batch =>
      batch.value().split("\n")
    }

    // 用正则匹配将日志格式化，并同时完成日期时间转时间戳
    val simpleDateFormat = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z", Locale.ENGLISH)

    val logRecords = lines.map { row =>
      val pattern: Regex = """^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$""".r
      val options = pattern.findFirstMatchIn(row)
      if (options.isDefined) {
        // 若匹配，返回格式化日志结构
        val matched = options.get
        log(matched.group(1), matched.group(2), matched.group(3),
          String.valueOf(simpleDateFormat.parse(matched.group(4)).getTime),
          matched.group(5), matched.group(6), matched.group(7), matched.group(8), matched.group(9))
      }
      else {
        // 若不匹配，用空值跳过
        log("foo", "foo", "foo", "0", "foo", "foo", "foo", "foo", "0")
      }
    }