Spark 사용

VPC 환경에서 이용 가능합니다.

Data Forest에서 Spark Job을 제출하는 방법과 Hive Metastore 연동 등의 추가 작업이 필요한 케이스에 대해서 설명합니다.

Spark Job 제출

Data Forest에서 사용자는 공용 Spark History Server를 사용하거나 개인용 Spark History Server를 실행시킬 수 있습니다.

Spark Job를 실행하려면 Dev 앱을 생성한 후, 원하는 Spark History Server 실행 방식을 정해 주십시오.

$ spark-submit \
 --master yarn \
 --queue longlived \
 ...
 --principal example@KR.DF.NAVERNCP.COM \
 --keytab df.example.keytab \ 
 --class com.naverncp.ExampleApplication \
 example_aplication.jar

Spark wordcount

텍스트 파일에서 각 단어의 개수를 반환하는 spark_wordcount.py로 Dev 셸(shell)에서 작업을 제출하는 예제입니다.
Spark History Server를 참조하여 Data Forest에서 생성한 히스토리 서버 앱을 연결했습니다.

1. Dev 셸에서 Kerberos 인증을 완료해 주십시오.

   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ kinit test01 -kt df.test01.keytab
   

2. spark_wordcount.py 프로그램을 다음과 같이 작성해 주십시오.

   import pyspark
   sc = pyspark.SparkContext()
   
   text_file = sc.textFile("input.txt")
   counts = text_file.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)
   counts.saveAsTextFile("output")
   

3. 작업을 제출하기 전에 이전에 생성된 output이 있다면 지워 주십시오.

   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ hdfs dfs -rm -r -f output
   21/04/30 14:49:56 INFO fs.TrashPolicyDefault: Moved: 'hdfs://koya/user/test01/output' to trash at: hdfs://koya/user/test01/.Trash/Current/user/test01/output
   

4. input.txt을 생성한 후 HDFS에 업로드해 주십시오.

   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ cp $SPARK_HOME/README.md input.txt
   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ hdfs dfs -put -f input.txt
   

   참고

   spark-submit 실행 시 --master, --deploy-mode를 지정하지 않으면 제출되지 않습니다. 주로 사용자의 애플리케이션은 worker machine과 물리적으로 먼 로컬 기기에서 제출되므로 driver과 executor 사이의 네트워크 지연이 생기게 됩니다. 이를 최소화 하기 위해 일반적으로 cluster mode를 사용합니다.

   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ spark-submit --py-files spark_wordcount.py \
   --master yarn \
   --deploy-mode cluster \
   spark_wordcount.py
   

5. 로그와 함께 작업 제출이 완료되면 다음과 같은 output의 내용을 확인할 수 있습니다.
   
6. Spark History Server에 접속하여 작업한 내용을 확인할 수 있습니다.
   

Spark SQL

Hive에 있는 규칙을 모두 지켜야 합니다. 단, Spark SQL에서는 규칙에 부합하지 않더라도 에러가 발생하지 않습니다.
예를 들어 사용자 'example'이 spark-sql에서 'example_db' 라는 이름으로 Hive 규칙에 어긋난 데이터베이스를 생성하는 경우, Beeline으로 조회 시 해당 데이터베이스는 조회되지 않습니다. 하지만 정기 감사를 통해 규칙에 어긋난 데이터베이스는 삭제하고 있으므로, Beeline에서 데이터베이스를 생성할 것을 권고합니다.

Hive 접속 설정

Spark SQL로 Hive 테이블에 접근할 수 있습니다. Spark에서 Hive Managed Table을 읽고 쓰려면 Hive Warehouse Connector(이하 HWC)가 필요하므로 테이블이 Managed인 경우, 추가 설정이 필요합니다.

Hive 테이블에 접근할 때 필요한 설정은 다음과 같습니다. Hive 테이블의 성격 (External/Managed)과 작업의 특성을 고려해서 필요한 옵션을 선택해 주십시오.

spark-submit을 통해 Batch Hive에 작업을 제출하는 예시는 다음과 같습니다. spark-shell, PySpark에도 동일한 설정을 추가할 수 있습니다.

$ spark-submit \
    --master yarn \
    --deploy-mode cluster \
    --queue longlived \
    ...
    --conf spark.sql.hive.hiveserver2.jdbc.url="jdbc:hive2://zk1.kr.df.naverncp.com:2181,zk2.kr.df.naverncp.com:2181,zk3.kr.df.naverncp.com:2181/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2" \
    --conf spark.datasource.hive.warehouse.metastoreUri="thrift://hms1.kr.df.naverncp.com:9083,thrift://hms2.kr.df.naverncp.com:9083" \
    --conf spark.hadoop.hive.zookeeper.quorum="zk1.kr.df.naverncp.com:2181,zk2.kr.df.naverncp.com:2181,zk3.kr.df.naverncp.com:2181"
    --principal example@KR.DF.NAVERNCP.COM \
    --keytab df.example.keytab \ 
    --jars /usr/hdp/current/hive_warehouse_connector/hive-warehouse-connector-assembly-1.0.0.3.1.0.0-78.jar \
    --class com.naverncp.ExampleApplication \
    example_aplication.jar

임의 Spark 버전으로 Spark 실행 환경 구성

Data Forest에서 제공하는 Spark 버전은 2.3.2입니다. 다른 버전의 Spark를 사용하고 싶은 경우, Hadoop Free 빌드 타입의 Spark를 설치합니다.

1. Spark (Hadoop Free) 다운로드

이 가이드에서는 2.4.7 버전을 설치하는 것을 기준으로 설명합니다.

Spark를 다운로드하는 방법은 다음과 같습니다.

1. Apache Spark 홈페이지에서 원하는 버전의 Spark를 다운로드해 주십시오.
2. 패키지 타입은 Pre-bulit with user-provided Apache Hadoop을 선택해 주십시오.

   $ mkdir -p $HOME/apps/spark
   $ wget -P $HOME/apps https://archive.apache.org/dist/spark/spark-2.4.7/spark-2.4.7-bin-without-hadoop.tgz
   $ tar xvfz $HOME/apps/spark-2.4.7-bin-without-hadoop.tgz -C $HOME/apps/
   $ ln -s $HOME/apps/spark-2.4.7-bin-without-hadoop $HOME/apps/spark
   $ SPARK_HOME=$HOME/apps/spark/spark-2.4.7-bin-without-hadoop
   $ SPARK_CONF_DIR=$HOME/apps/spark/spark-2.4.7-bin-without-hadoop/conf
   

2. Configuration 설정

1. 아래의 기본값 설정을 먼저 복사해 주십시오.

   $ cp /etc/spark2/conf/* $SPARK_CONF_DIR/
   

2. $SPARK_CONF_DIR/spark-defaults.conf 파일에 아래 다음과 같이 변수를 추가해 주십시오.

   spark.driver.extraJavaOptions -Dhdp.version=3.1.0.0-78
   spark.yarn.am.extraJavaOptions -Dhdp.version=3.1.0.0-78
   
   spark.yarn.appMasterEnv.YARN_CONTAINER_RUNTIME_DOCKER_MOUNTS /usr/hdp:/usr/hdp:ro
   spark.executorEnv.YARN_CONTAINER_RUNTIME_DOCKER_MOUNTS /usr/hdp:/usr/hdp:ro
   

3. $SPARK_CONF_DIR/spark-env.sh 파일에 아래와 같이 추가해 주십시오.

   export SPARK_SUBMIT_OPTS="-Dhdp.version=3.1.0.0-78"
   export PATH=$SPARK_HOME/bin:$PATH
   export SPARK_DIST_CLASSPATH=`$HADOOP_COMMON_HOME/bin/hadoop classpath`
   

3. jars 업로드

Data Forest 에서 제공하는 hadoop-*.jar에는 여러 종류의 버그 패치가 적용되어 있습니다. 기타 사용자가 필요한 jar을 해당 경로 아래 복사해 두면 jar 업로드를 위한 추가 옵션(extraClassPath)을 설정하지 않아도 됩니다.

jars를 업로드하는 방법은 다음과 같습니다.

1. $SPARK_HOME/jar아래 해당 jar 파일들을 복사해 주십시오.

   cp /usr/hdp/current/spark2-client/jars/hadoop-*.jar $SPARK_HOME/jars
   

2. $SPARK_HOME/jar를 압축해 주십시오.

   cd $SPARK_HOME/jars
   tar cvfz /tmp/spark_jars.tar.gz *
   

3. HDFS 사용자 홈 디렉터리 아래에 파일을 업로드해 주십시오.

   kinit example@KR.DF.NAVERNCP.COM -kt example.service.keytab
   hadoop fs -copyFromLocal /tmp/spark_jars.tar.gz /user/example/
   hadoop fs -setrep 10 /user/example/spark_jars.tar.gz
   

4. spark-submit 실행

spark-submit 실행 시 spark.yarn.archive 옵션을 붙여서 실행해 주십시오.
아래 예제에서는 위에서 작성했던 spark_wordcount.py와 input.txt 파일을 그대로 사용합니다.

$SPARK_HOME/bin/spark-submit --master yarn spark_wordcount.py # 기타 옵션 ... 
--conf spark.yarn.archive=hdfs://koya/user/example/spark_jars.tar.gz \

Spark 3.0.0 이상을 사용하는 경우, 사용하려는 HDFS 네임스페이스를 지정하는 옵션을 하나 더 추가해 주십시오.

--conf spark.kerberos.access.hadoopFileSystems=hdfs://koya,hdfs://tata