Python: Spark-Kafka-Sql(5)
# 활용 # 파일다운 # 스파크 # 카프카 # SQL
드디어 활용을 해봄.
스파크 카프카(Structured Streaming SQL) 파이썬.
조합이 좋음.
시나리오는
Suricata 를 통해서 나오는 네트워크 트래픽 중에서
File 관련 이벤트 중에서
PDF, EXE, MS 관련 파일들의 이름을
가져와서 보여주는데
이걸 실시간으로!
참고로 Suricata 에서 나오는 로그는 초당 3000개 정도.
초당 3000개 중에서 내 IP, 파일 이벤트, 확장자 를 필터로 해서 원하는 정보를 가져옴.
SQL을 간단히 보여주면.
selectExpr("data.flow_id as flow_id", "data.src_ip as src_ip", "data.dest_ip as dest_ip", "data.event_type as event_type", "data.fileinfo.md5 as md5", "data.fileinfo.sha256 as sha256",
"lower(data.fileinfo.magic) as magic", "data.fileinfo.filename as fname", "data.http.hostname as hname", "timestamps as ts") \
.where('event_type = "fileinfo"') \
.where('dest_ip like "%192.168.2.11" or src_ip like "%192.168.2.11"') \
.where(' magic like "%pdf%" or magic like "%micro%" or magic like "%zip%" ') \
.groupBy('magic', 'fname', window('ts', '1000 minutes').alias('windows'))\
.agg(max('event_type').alias('event_type'), max('flow_id').alias("flow_id"), max('ts').alias("ts"), count('magic').alias('magic_cnt'), count('fname').alias('fname_cnt')) \
.orderBy("ts", ascending=False)
간단하지가 않네.
다행히 오라클 SQL 교육을 전직장에서 빡시게 받아서
쿼리는 좀 볼 줄 알게 되었음.
위의 쿼리는 위위에서 설명한 시나리오를 옮긴 것.
저 쿼리는 스파크에서 돌아가면서 리얼타임으로 결과를 쏟아냄.
일단 동영상으로 보면.
결과가 다 나왔음.
다시 설명하자면, 내 IP로 다운받은 파일에 대해서만 추출해 낸 것.
초당 3000~5000 정도의 이벤트를 실시간으로 처리. 빠르게.
끝.
