[Spark] RDD의 사용 실습 & HDFS의 웹 로그 데이터 삽입(Flume)

 

Spark에서 작업하기 전에 해야 할 일

1. 웹 로그 데이터를 HDFS에 넣기(Flume이 해줍니다.)
2. Flume이 "로컬 로그 파일"을 읽어서 HDFS의 "/loudacre/weblogs" 디렉토리로 넣는 과정을 수행해야 하겠습니다.
3. Spark는 HDFS에 있는 데이터를 기반으로 작업하기 때문에, 그 데이터가 먼저 존재해야 Spark 코드가 제대로 실행되기 때문입니다.
4. 이 실습에서는 제공된 웹 로그 파일을 로컬 스풀(spool) 디렉토리에 배치하여 "Apache 서버"를 시뮬레이션한 후, "Flume"을 사용하여 데이터를 수집합니다.

 

Flume이 수집한 데이터를 저장할 HDFS 디렉토리 생성

$ hdfs dfs -mkdir /loudacre/weblogs

 

웹 서버 로그 출력을 위한 로컬 디렉토리 생성, 모든 사용자가 해당 디렉토리에 쓰기 권한을 가지도록 설정

$ sudo mkdir -p /flume/weblogs_spooldir
$ sudo chmod a+w -R /flume

 

$DEV1/exercises/flume 디렉토리에 다음과 같은 특성을 가진 "Flume 구성 파일"을 생성하고 "Flume 에이전트" 실행

$ cd $DEV1/exercises/flume
$ flume-ng agent --conf /etc/flume-ng/conf \
  --conf-file spooldir.conf \
  --name agent1 –Dflume.root.logger=INFO,console

 

Apache 웹 서버 출력 시뮬레이션

 

새로운 "터미널 창"을 열고, 실습 디렉토리로 이동한 후, 웹 로그 파일을 "/flume/weblogs_spooldir"로 복사하는 스크립트를 실행합니다.( 이 스크립트는 웹 로그 파일의 임시 복사본을 만들고 스풀 디렉토리로 이동시킵니다.)

$ cd $DEV1/exercises/flume
$ ./copy-move-weblogs.sh /flume/weblogs_spooldir

1. Flume 에이전트가 실행되는 터미널로 돌아가 "로그 출력"을 확인하세요.
2. Flume이 "HDFS에 데이터를 저장하는 정보"를 볼 수 있습니다.

.tmp 확장자를 제거하고, 최종 파일명으로 변경확인

 

RDD 사용 실습

Python Spark Shell 시작하기

$ pyspark

 

SparkContext 객체 확인

pyspark> sc

 

Spark Shell 종료

• Ctrl + D
• exit 입력 후 Enter

RDD 생성 (Python 또는 Scala, 저는 Python기준으로 합니다.)

pyspark> mydata = sc.textFile("file:/home/training/training_materials/data/frostroad.txt")

 

데이터 개수 확인 (RDD 연산)

pyspark> mydata.count()

 

 

Spark Shell 종료

pyspark> exit

 

RDD를 사용하여 데이터셋 변환하기

• 실습 디렉터리: $DEV1/exercises/spark-transform
• 데이터 파일: /loudacre/weblogs/* (HDFS)

이번 실습에서는 Spark Shell에서 RDD를 사용하는 연습을 합니다.

이전에 Flume에서 수집한 웹 서버 로그 데이터를 Spark로 알아보는 과정입니다.

 

데이터 파일 경로를 변수에 저장하여 반복 입력을 방지하기(logfile)

$ pyspark
pyspark> logfile="/loudacre/weblogs/FlumeData.*"

 

RDD 생성 및 변환

데이터 파일을 RDD로 생성(logfile)

pyspark> logs = sc.textFile(logfile)

데이터 파일을 RDD로 생성

map() 함수를 사용하여 새로운 RDD를 생성하세요.

pyspark> logs.map(lambda line: len(line)).take(5)

pyspark> logs.map(lambda line: line.split()).take(5)

 

로그 파일에서 IP 주소만 포함하는 RDD를 만들고 출력

pyspark> ips = logs.map(lambda line: line.split()[0])
pyspark> ips.take(5)

 

IP 주소 데이터를 더 읽기 쉽게 출력하기

pyspark> for ip in ips.take(10): print ip