Data Forest에서 Spark, Hive로 데이터 처리하기 Part.1
네이버 클라우드 플랫폼 Data Forest는 빅데이터 처리를 위한 다양한 오픈소스 프레임워크를 제공합니다. 이번 페이지에서는 Data Forest에서 제공하는 Hadoop, Spark, Hive를 이용하여 데이터를 처리하고 저장하는 예제를 수행해보겠습니다. Spark로 영화정보와 평점데이터 셋 MovieLens 20M 을 처리하여 HDFS에 결과를 저장하고, 그 데이터로 Hive 테이블을 만들어보겠습니다.
이번 페이지에서는 모든 처리를 Data Forest App에서 생성한 Zeppelin 노트북에서 진행하려고 합니다.
Spark Job실행은 spark-submit , spark-shell (REPL) 같은 스크립트를 사용할 수 있습니다. Hive 쿼리 또한 beeline 같은 클라이언트를 실행해도 됩니다.
이후 Part. 2에서는 DEV앱을 이용하여 spark-submit beeline 커맨드로 동일한 작업을 하는 방법도 소개하겠습니다.
Data Forest App 생성하기
Zeppelin앱은 Apache Zeppelin 서버를 생성합니다. 네이버 클라우드 플랫폼(ncloud.com) 콘솔 [Data Forest > App] 에서 + App 생성을 클릭하여 앱을 생성할 수 있습니다.
앱 타입은 ZEPPELIN-0.8.1을 선택합니다.
앱이 STABLE 상태가 되면 생성이 완료된 것 입니다.
데이터 처리하기
먼저 원본 데이터셋을 업로드 합니다. 실제 서비스에서는 Kafka, Flume등을 통해서 원본 로그 데이터 등을 HDFS에 저장하거나, RDBMS로 부터 Sqoop을 사용하여 원본 데이터를 HDFS에 보관할 수 있습니다. 여기서는 앞서 언급한 MovieLens 데이터셋의 .csv 파일을 HDFS에 직접 업로드 하도록 하겠습니다.
HDFS 네임노드 UI 에 접속하여 /user/${USER} 디렉토리로 이동합니다. Upload 버튼을 클릭해서 rating.csv , movie.csv 파일을 업로드 합니다. 참고로 업로드 완료 된 파일을 클릭한 뒤 Head the file을 해보면 데이터가 어떻게 생겼는지 볼 수 있습니다.
1. Zeppelin 앱 접속하기
Zeppelin에 접속합니다. URL은 Zeppelin 앱의 [Quick links > zeppelin] 을 참고합니다.
계정을 생성할 때 지정한 비밀번호를 사용하여 로그인 하면 됩니다.
2. Spark Job 수행하기
Notebook > +Create new note로 새로운 노트북을 생성합니다. Default Interpreter는 Spark로 지정합니다. Spark 코드에는 %spark 를 붙여서 인터프리터를 실행 시킵니다.
Spark로 두 개의 .csv 파일로 부터 데이터프레임을 생성하고, 두 개의 데이터프레임을 조인하여 Parquet파일로 HDFS에 저장해 보겠습니다. 해당 Zeppelin 노트북의 전체 내용은 여기에서 다운로드 하여 Import할 수 있습니다.
movie.csv , rating.csv 를 그대로 쓰는222 것이 아니라 새로운 컬럼을 만들고 변경하는 등 기본적인 transformation 작업을 수행합니다.
위에서 등록한 view에 대해서 spark-sql을 실행 할 수 있습니다. 아래 쿼리는 단순한 예시입니다.
두개의 데이터프레임으로부터 새로운 데이터프레임을 생성합니다.
만들어진 최종 데이터프레임을 Parquet형식으로 저장합니다. year, month을 파티션으로 지정합니다.
쓰기 작업이 완료되면, HDFS 해당 경로에 아래 처럼 파티션이 생성 되고 데이터가 써진 것을 확인 할 수 있습니다.
진행이 끝난 작업은 Spark History Server에서 확인 할 수 있습니다. 진행중인 작업은 Show Incomplete Applications 버튼을 클릭하면, 확인 할 수 있습니다.
3. Hive 쿼리 수행하기
JDBC 인터프리터 사용 전 한 가지 설정이 필요합니다. Data Forest에서 제공하는 HiveServer2 는 계정명/패스워드로 인증을 해야 접근할 수 있는데, 기본 인터프리터에는 패스워드 설정이 들어가 있지 않으므로 사용자가 직접 추가해주어야 합니다.
패스워드 설정을 하지 않고 쿼리를 수행하면 org.apache.zeppelin.interpreter.InterpreterException: Error in doAs 가 발생하므로 주의합니다.
우측 상단 계정명 > Interpreter 에서 jdbc interpreter를 찾았으면 edit 버튼을 클릭하여 아래 처럼 hive.password 항목을 추가합니다. 값은 Data Forest Account를 생성했을 때 지정한 패스워드 입니다. Save를 눌러 변경사항을 저장하고 인터프리터를 재시작 합니다.
Notebook > +Create new note로 새로운 노트북을 생성합니다. Default Interpreter는 jdbc로 지정합니다. Hive 쿼리에는 %jdbc(hive) 를 붙여서 인터프리터를 실행 시킵니다.
Spark Job에서 쓴 Parquet 파일로 부터 Hive 테이블을 만들고, 쿼리를 실행해 보겠습니다. 해당 Zeppelin 노트북의 전체 내용은 여기에서 다운로드 하여 Import할 수 있습니다.
External 데이터베이스를 만들 때 LOCATION 키워드로 경로를 항상 정해줘야 합니다. Data Forest 사용자는 Hive default 데이터베이스에 접근할 수 없기 때문에, 항상 자신의 HDFS 홈 디렉토리 ( /user/${USER} )아래 경로를 사용해야 합니다. 그리고 Database 이름을 설정하는것에도 규칙이 있습니다.
항상 ${USER}__db_${ANY_NAME_YOU_WANT} 형식으로 데이터베이스를 생성해야 하며, 그렇지 않으면 오류가 발생합니다.
데이터베이스와 달리, 테이블 생성시 LOCATION 키워드는 옵션입니다. 여기서는 기존에 있던 파일로부터 테이블을 생성할 것이기 때문에, LOCATION 뒤에 경로로 Parquet 파일이 저장된 디렉토리를 설정했습니다.
MSCK REPAIR TABLE ${TALBENAME} 으로 파티션 메타데이터를 추가해줍니다.
이제 Hive 테이블을 사용할 준비가 끝났습니다. 사용자가 원하는대로 쿼리를 수행 할 수 있습니다. 아래 쿼리는 단순한 예시입니다.
수행한 Hive 쿼리에 대한 작업 진행사항은 Resource Manager UI에서 확인 할 수 있습니다. Log를 클릭해서 작업 로그를 볼 수 있습니다.
마치며
Zeppelin은 사용자가 쉽게 API 코드를 작성하고, 제출 할 수 있는 환경을 제공합니다. 다양한 인터프리터가 있지만 가장 사용 빈도가 높은 Spark, Hive를 사용해 봤습니다.
Zeppelin 안에서 Spark 로 데이터를 처리하고 그 결과를 HDFS에 저장한 다음 다시 Hive 테이블로 읽어오는 작업을 수행해 보았는데요. 이 예제를 응용하면, Data Forest 플랫폼 위에서 스트리밍 데이터소스에 Spark Streaming을 사용하여 HDFS에 데이터를 저장하고 주기적으로 Hive쿼리 배치를 수행하는 작업을 만들 수 있습니다.
다음 글 Part.2 에서는 Zeppelin 노트북이 아니라 커맨드 라인으로 스크립트를 사용하여 Spark Job을 제출하고 Hive쿼리를 실행하는 방법을 알아보겠습니다.
