java - 如何使用 Spark 在 Java 中映射日志文件？

Question

我必须监视一个日志文件，其中写入了应用程序的使用历史记录。该日志文件的格式如下:

<AppId,date,cpuUsage,memoryUsage>
<AppId,date,cpuUsage,memoryUsage>
<AppId,date,cpuUsage,memoryUsage>
<AppId,date,cpuUsage,memoryUsage>
<AppId,date,cpuUsage,memoryUsage>
... about 800000 rows

AppId 始终相同，因为仅在一个应用中引用，date 以此格式表示 dd/mm/yyyy hh/mm cpuUsage 和 memoryUsage 以 % 表示，例如:

<3ghffh3t482age20304,230720142245,0.2,3,5>

具体来说，我必须检查该应用程序的 CPU 使用率和内存使用率，以使用 Spark 和 MapReduce 算法进行监控。

我的输出是当 cpu 或内存使用 100% 时打印警报。

我该如何开始？

Answer 1

这个想法是声明一个类并将该行映射到一个 scala 对象中，

让我们按如下方式声明案例类，

case class App(name: String, date: String, cpuUsage: Double, memoryusage: Double)

然后初始化 SparkContext 并从存在数据的文本文件创建一个 RDD，

val sc = new SparkContext(sparkConf)
val inFile = sc.textFile("log.txt")

然后解析每一行并将其映射到App对象，这样范围检查会更快，

val mappedLines = inFile.map(x => (x.split(",")(0), parse(x)))

其中 parse(x) 方法定义如下，

 def parse(x: String):App = {
   val splitArr = x.split(",");
   val app = new App(splitArr(0),
                      splitArr(1),
                      splitArr(2).toDouble,
                      splitArr(3).toDouble)
   return app
}

请注意，我假设输入如下，(这只是为了给您提供想法，而不是整个程序)，

ffh3t482age20304,230720142245,0.2,100.5

然后进行过滤器转换，您可以在其中执行检查并报告异常条件，

val anamolyLines = mappedLines.filter(doCheckCPUAndMemoryUtilization)
anamolyLines.count()

其中doCheckCPUAndMemoryUtilization函数定义如下，

def doCheckCPUAndMemoryUtilization(x:(String, App)):Boolean = {
    if(x._2.cpuUsage >= 100.0 ||
       x._2.memoryusage >= 100.0) {
       System.out.println("App name -> "+x._2.name +" exceed the limit")
       return true
    }

    return false
}

注意:这只是批处理，不是实时处理。