java - 如何使用 Java 在 Spark 中处理多行输入日志文件

Question

我是 Spark 的新手，对我来说似乎很困惑。我已经浏览了 Java API 的 spark 文档，但无法找到解决问题的方法。我必须在 spark-Java 中处理一个日志文件并且剩下的时间很少。下面是包含跨多行的设备记录(设备 ID、描述、IP 地址、状态)的日志文件。它还包含一些我不关心的其他日志信息。如何从这个巨大的日志文件中获取设备信息日志。非常感谢任何帮助。

输入日志数据:

!
!

!  
 device AGHK75  
  description "Optical Line Terminal"  
  ip address 1.11.111.12/10     
  status "FAILED"  
!  
 device AGHK78  
  description "Optical Line Terminal"  
  ip address 1.11.111.12/10     
  status "ACTIVE"  
!  

!  
context local  
!  
 no ip domain-lookup  
!  
 interface IPA1_A2P_1_OAM  
  description To_A2P_1_OAM  
  ip address 1.11.111.12/10     
  propagate qos from ip class-map ip-to-pd  
!  
 interface IPA1_OAM_loopback loopback  
  description SE1200_IPA-1_OAM_loopback  
  ip address 1.11.111.12/10     
   ip source-address telnet snmp ssh radius tacacs+ syslog dhcp-server tftp ftp    icmp-dest-unreachable icmp-time-exceed netop flow-ip 

到目前为止我所做的是:
Java 代码

JavaRDD<String> logData = sc.textFile("logFile").cache();
List<String> deviceRDD = logData.filter(new Function<String, Boolean>() {
Boolean check=false;
public Boolean call(String s) {         
    if(s.contains("device") ||(check == true && ( s.contains("description") || s.contains("ip address"))))
        check=true;
    else if(check==true && s.contains("status")){
        check=false;
        return true;
        }
    else
        check=false;
    return check; }
}).collect();

当前输出:

device AGHK75
description "Optical Line Terminal"   
ip address 1.11.111.12/10   
status "FAILED"   
device AGHK78   
description "Optical Line Terminal"   
ip address 1.11.111.12/10   
status "ACTIVE"  

预期输出是:

AGHK75,"Optical Line Terminal",1.11.111.12/10,"FAILED"   
AGHK78,"Optical Line Terminal",1.11.111.12/10,"ACTIVE"

Answer 1

您可以使用 sc.wholeTextFiles("logFile") 获取数据作为键，值对，其中键是文件名，值是其中的数据。

然后您可以使用一些字符串操作根据带有“!”的单个日志数据的开始和结束分隔符来拆分数据。并首先做一个过滤器来检查第一个词是否是设备，然后在它上面做一个 flatMap ，这将使它成为 singleLog 文本 RDD。

然后使用 map 从中获取数据。

请尝试一下，让我知道这个逻辑是否适合您。

在 Spark Scala 中添加代码:

val ipData = sc.wholeTextFiles("abc.log")
val ipSingleLog = ipData.flatMap(x=>x._2.split("!")).filter(x=>x.trim.startsWith("device"))
val logData = ipSingleLog.map(x=>{
  val rowData = x.split("\n")
  var device = ""
  var description = ""
  var ipAddress = ""
  var status = ""
  for (data <- rowData){
    if(data.trim().startsWith("device")){
      device = data.split("device")(1)
    }else if(data.trim().startsWith("description")){
      description = data.split("description")(1)
    }else if(data.trim().startsWith("ip address")){
      ipAddress = data.split("ip address")(1)
    }else if(data.trim().startsWith("status")){
      status = data.split("status")(1)
    }
  }
  (device,description,ipAddress,status)
})
logData.foreach(println)