时间戳

首页 > 解决方案 > 如何使用 spark 过滤流数据并将其从 apache flume 转换为 rdd/data freame 以将其写入表

apache-spark - 如何使用 spark 过滤流数据并将其从 apache flume 转换为 rdd/data freame 以将其写入表

问题描述

嗨，我是水槽/Spark/Spark 流媒体的新手。我已经配置了水槽和 netcat 并成功地将数据流式传输到 Spark。

我的要求是从日志文件中检查流数据（水槽流）中的错误并获取错误行（流中出现的行中的单词“ERROR”）并将其作为 DF 将其写入 oracle。

我在下面的过滤器中遇到异常并转换为 DF 代码。请帮我解决这个问题

import org.apache.spark.streaming.flume.FlumeUtils
import org.slf4j.LoggerFactory
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{StreamingContext,Seconds}
import org.apache.spark.streaming.flume._
import org.apache.spark._
import org.apache.spark.streaming._
import spark.implicits._
val hostName = "10.90.3.78"
val port = 9999.toInt
val sparkStreamingContext = new StreamingContext(sc,Seconds(10))
val stream = FlumeUtils.createPollingStream(sparkStreamingContext,hostName,port)
val mappedlines = stream.map( e => new String(e.event.getBody.array()))
.filter(rec => rec.contains("ERROR"))
.map(line => line.split("ERROR"))
val arr = mappedlines.foreachRDD({status=>val DF = status.toDF()})
println(arr)
stream.count().map(cnt => "Received " + cnt + " flume events." ).print()
sparkStreamingContext.start()
sparkStreamingContext.awaitTermination()

标签： apache-sparkapache-spark-sqlspark-streamingflume

解决方案

我已经使用 Foreach 解决了它并将 RDD 转换为 DF。它起作用了，我已成功将错误行插入到 DB 中。

推荐阅读