Flink/SparkのELTの簡単なテンプレート
われわれは往々にして外部の妨害要素を無視し、焦りを避け、自分のやりたいことに専念することができる.どうせ焦りは問題を解決できない.
引用する
flinkやsparkを使うときは、固定的なテンプレートを書くことが大切です.etlのリアルタイムタスクについては、mapとfliterを1つ実行するだけで解決します.簡単に抽象化して、パラメータを着ることができます.
親
etl
反射方法(Reflection Method)
使用
StageDataOperationPushLogのフルパス
引用する
flinkやsparkを使うときは、固定的なテンプレートを書くことが大切です.etlのリアルタイムタスクについては、mapとfliterを1つ実行するだけで解決します.簡単に抽象化して、パラメータを着ることができます.
親
abstract class BashLogLineParse extends Serializable{
var line: String= ""
def logParse():String
def logFilter(mes:String):Boolean
}
etl
import java.text.SimpleDateFormat
import java.util.{
Date, Locale}
import com.alibaba.fastjson.JSON
import com.oneniceapp.base.BashLogLineParse
import org.apache.commons.lang3.time.FastDateFormat
class StageDataOperationPushLog extends BashLogLineParse with Serializable{
private val YYYYMMDDHHMM_TIME_FORMAT = FastDateFormat.getInstance("dd/MMM/yyyy:HH:mm:ss Z", Locale.ENGLISH)
override def logParse(): String = {
val line = this.line
val log = JSON.parseObject(line)
val datastr = log.getString("raw")
val data = JSON.parseObject(datastr)
val payloadstr = data.getString("payload")
val payload = JSON.parseObject(payloadstr)
val notice_id = payload.getString("notice_id")
val user_id = data.getString("receiver")
val raw_time = data.getLong("raw_time")
val partition_date = new SimpleDateFormat("yyyy-MM-dd").format(new Date(raw_time*1000))
data.remove("receiver")
data.put("user_id", user_id)
data.put("notice_id", notice_id)
data.put("partition_date", partition_date)
if(notice_id==null)
""
else
data.toJSONString
}
override def logFilter(mes:String):Boolean ={
if(!mes.equals(""))
true
else
false
}
}
反射方法(Reflection Method)
public class CreateObject {
public static Object createInstance(String className){
try {
Class clz = Class.forName(className);
Object obj = clz.newInstance();
return obj;
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}
}
使用
StageDataOperationPushLogのフルパス
mapDate = transction.map(line=>{
val ls = CreateObject
.createInstance(conf.classFile)
.asInstanceOf[BashLogLineParse]
ls.line = line.message
(line.topic, line.partition, line.offset, ls.logParse())
})
.filter(line=>{
val ls = CreateObject
.createInstance(conf.classFile)
.asInstanceOf[BashLogLineParse]
ls.logFilter(line._4)
})