0
さまざまな種類のログメッセージで満たされたテキストファイルがたくさんありますが、ファイルごとに複数の種類のメッセージを表示することはできません。異なるフォーマットのテキストファイルをデータセットにマップ
ファイル1：（I）; 2017年1月12日; 16：54：45;ランダムなテキスト、他のテキスト
ファイル2： （I）; 2017年1月13日15：34：56;アゲインテキスト;もう1つの時間// ieスペース 日付と時刻の間
私はすでにこれを動作させていますが、これが正しい方法であるかどうか尋ねたいと思います。また、私の方法は、セミコロンとスペースの間の変更が常に同じ位置に現れる場合にのみ機能します。
私はscala/sparkを初めて使用しているので、この点に関するアドバイスをいただければ幸いです。
//read file
val df = spark.read.textFile(file.path).filter(f => f.nonEmpty && f.length > 1 && f.startsWith("("))
//create empty dataset of type OutputMessage
var df3 = Seq.empty[OutputMessage].toDS()
//get number of semicolons within first line of the dataset to determine type
val message_type = df.take(1).mkString(",").count(_ == ';')
if(message_type == 5){
//split by semicolon and create dataset of type InputMessage
var df2 = df.map(x => x.split(";")).map(x => InputMessage(x(0), x(1), x(2), x(3), x(4), x(5)))
//map to dataset of type output message
df3 = df2.map(
x =>
OutputMessage(x.status,
x.messages_datestring,
x.messages_timestring,
x.device,
x.device_fullmessage,
x.device_message,
fileName,
getWeekday(x.messages_datestring),
(x.messages_datestring + "T" + x.messages_timestring),
data_company,
data_location,
data_systemname)
)
}
else if (message_type == 4){
var df2 = df.map(x => x.split(";")).map(x => InputMessage1(x(0), x(1), x(2), x(3), x(4)))
df3 = df2.map(
x=>
OutputMessage(x.status,
x.messages_datetimestring.split(" ").take(1).mkString(","),
x.messages_datetimestring.split(" ").takeRight(1).mkString(","),
x.device,
x.device_fullmessage,
x.device_message,
fileName,
getWeekday(x.messages_datetimestring.split(" ").take(1).mkString(",")),
x.messages_datetimestring.replace(' ', 'T'),
data_company,
data_location,
data_systemname)
)
}
//convert to rdd
val dsToRDD = df3_filtered.rdd
//laod to elasticsearch
dsToRDD.saveToEs("abdata/log")
EDIT：一部のファイルには行間に不一致があることがわかりました。つまり、私の解決策はもはや実際には適用できません。
EDIT：ラインベースの実行に変更されました。行の中のランダムな分離を除いて、ほとんどのものはこれまでのところ動作します。私はこのケースのための出力を得るが、望んでいない。
object MapRawData{
def mapRawLine (line: String): Option[RawMessage] ={
var msgtype = 0;
val fields = line.split(";")
if (fields(0).length == 3 && fields(1).length == 10) msgtype = 1
if (fields(0).length == 3 && fields(1).length > 10) msgtype = 3
if (fields(0).length > 16) msgtype = 2
try {
fields.map(_.trim)
Some(
RawMessage(
status = fields(0).take(3),
messages_datestring = if(msgtype == 1) fields(1) else if(msgtype == 2) fields(0).drop(4).take(10) else fields(1).take(10),
messages_timestring = if(msgtype == 1) fields(2).take(8) else if (msgtype == 2) fields(0).drop(15).take(8) else (fields(1).drop(11).take(8)),
device = if(msgtype == 1) fields(3) else if (msgtype == 2) fields(1) else fields(2),
device_fullmessage = if(msgtype == 1) fields(4) else if (msgtype == 2) fields(2) else fields(3),
device_message = if(msgtype == 1) fields(5) else if (msgtype == 2) fields(3) else fields(4)
)
)
}
catch {
case e: Exception =>
println(s"Unable to parse line: $line")
None
}
}
}
この変化は最初のものよりも時間がかかりますか？