Zeppelin + Spark + Hive LLAP

이 문장에 대하여

이 포스트는, 요전날 Hadoop과 S3 라고 하는 타이틀로 쓴 아래와 같은 아키텍쳐를 만들기에 있어서의 Zeppelin + Spark + LLAP 주위의 조사와 검증 상황을 자신을 위한 메모로서 정리한 것입니다.

Motivation

Spark에서 S3의 데이터를 분석할 때 대기 시간을 가능한 한 낮추고 S3의 스로틀링을 피하기 위해 가능한 한 액세스 빈도를 낮추고 싶다는 배경에서 spark-llap을 시도하고 있다.

spark-llap은 단순히 Hiveserver2 경유로 Hive LLAP의 실행 결과를 받는 것은 아니고, 다음과 같은 아키텍처로 데이터를 꺼낸다.



* 심상은 Hortonworks Hive 개발자, Sergey Shelulkhin의 Slideshare 에서 빌림.

이 방법의 장점은 Hiveserver2에 집계되어 Thrift에 직렬화된 결과를 받는 것이 아니라 Spark의 각 Executor가 직접 LLAP Daemon에서 데이터를 꺼내게 되므로 병목이 없다는 것이다.

빌드

README에는 build/sbt package 제가 쓰여져 있지만 이것으로 안 되고 아래와 같이 build/sbt assembly 해야 한다. (글쎄, 그렇지?)

master 브랜치는 Spark1.6용이므로 이런 느낌으로 빌드할 수 있다.

build/sbt -Dspark.version=1.6.2.2.5.3.0-37 -Dhadoop.version=2.7.3.2.5.3.0-37 -Dhive.version=2.1.0.2.5.3.0-37 -Dtez.version= 0.8.4.2.5.3.0-37 -Drepourl= ㅜㅜㅜㅜㅜㅜㅜㅜ rrun을 rks. 코 m : 8081 / 네오 s / 혼텐 t / g 납 ps / 푸 b ぃc / clean compile assembly

Spark2.0용 branch-2.0도 아래와 같은 느낌으로 빌드.

build/sbt -Dspark.version=2.0.0.2.5.3.0-37 -Dhadoop.version=2.7.3.2.5.3.0-37 -Dhive.version=2.1.0.2.5.3.0-37 -Dtez.version= 0.8.4.2.5.3.0-37 -Drepourl= ㅜㅜㅜㅜㅜㅜㅜㅜ rrun을 rks. 코 m : 8081 / 네오 s / 혼텐 t / g 납 ps / 푸 b ぃc / clean compile assembly

둘 다 잘 갔다. Hortonworks의 repository를 사용하지 않으면 안 되는지는 ... 때문에 하고 있지 않다.

움직여 보자

README에 따라 다음을 spark-default.conf로 설정

spark.sql.hive.hiveserver2.url=jdbc:hive2://hiveserver2:10500
spark.hadoop.hive.llap.daemon.service.hosts=@llap0
spark.hadoop.hive.zookeeper.quorum=zookeeper:2181

Spark Thrift Server 사용한다면 그 밖에도 설정이 필요하지만 여기에서는 스킵. 여기에서 spark-shell --jars PATH/TO/ASSEMBRY_JAR 에서 spark-shell을 시작합니다. 아래의 명령으로 움직여 주었다.

import org.apache.spark.sql.hive.llap.LlapContext
var llapContext = new LlapContext(sc)

val sql = "SOME HIVE QL"

llapContext.sql(sql).show

Spark2라면 아래에서만 움직이는 것 같다. 쿼리는 통과하지만 실제로 LLAP에서 작업이 실행 중인지 여부

spark.sql("SOME HIVE QL").show

제대로 움직이고 있는지 확인

Spark1에서의 실행은 여러가지 디버그 메시지가 나오므로 움직이고 있는 것 같지만, Spark2의 쪽은 지금 디버깅 방법을 모른다.

Spark에서 움직이고 있기 때문에 (아마), Tez View에는 로그가 나오지 않고 LLAP Web Service (http://llapnode:10502)에도 어떤 정보도 나오지 않는다

Zeppelin으로 움직여보세요

Zeppelin에서 Spark Interpreter에 spark-llap의 assembly jar를 더하면 다음 오류가 발생합니다.

import org.apache.spark.sql.hive.llap.LlapContext
java.lang.RuntimeException: java.lang.RuntimeException: Unable to instantiate org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient
    at org.apache.hadoop.hive.ql.session.SessionState.start(SessionState.java:522)
    at org.apache.spark.sql.hive.client.ClientWrapper.<init>(ClientWrapper.scala:204)
    at org.apache.spark.sql.hive.client.IsolatedClientLoader.createClient(IsolatedClientLoader.scala:238)
    at org.apache.spark.sql.hive.HiveContext.executionHive$lzycompute(HiveContext.scala:225)
    at org.apache.spark.sql.hive.HiveContext.executionHive(HiveContext.scala:215)
    at org.apache.spark.sql.hive.HiveContext.setConf(HiveContext.scala:458)
    at org.apache.spark.sql.SQLContext$$anonfun$4.apply(SQLContext.scala:272)
    at org.apache.spark.sql.SQLContext$$anonfun$4.apply(SQLContext.scala:271)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at org.apache.spark.sql.SQLContext.<init>(SQLContext.scala:271)
    at org.apache.spark.sql.hive.HiveContext.<init>(HiveContext.scala:90)
    at org.apache.spark.sql.hive.llap.LlapContext.<init>(LlapContext.scala:40)
    at org.apache.spark.sql.hive.llap.LlapContext.<init>(LlapContext.scala:60)
    at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:30)
    at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:36)
    at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:38)
    at $iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:40)
    at $iwC$$iwC$$iwC$$iwC.<init>(<console>:42)
    at $iwC$$iwC$$iwC.<init>(<console>:44)
    at $iwC$$iwC.<init>(<console>:46)
    at $iwC.<init>(<console>:48)
    at <init>(<console>:50)
    at .<init>(<console>:54)
    at .<clinit>(<console>)
    at .<init>(<console>:7)
    at .<clinit>(<console>)
    at $print(<console>)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:1065)
    at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1346)
    at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:840)
    at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:871)
    at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:819)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.apache.zeppelin.spark.Utils.invokeMethod(Utils.java:38)
    at org.apache.zeppelin.spark.SparkInterpreter.interpret(SparkInterpreter.java:717)
    at org.apache.zeppelin.spark.SparkInterpreter.interpretInput(SparkInterpreter.java:928)
    at org.apache.zeppelin.spark.SparkInterpreter.interpret(SparkInterpreter.java:871)
    at org.apache.zeppelin.spark.SparkInterpreter.interpret(SparkInterpreter.java:864)
    at org.apache.zeppelin.interpreter.LazyOpenInterpreter.interpret(LazyOpenInterpreter.java:94)
    at org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:341)
    at org.apache.zeppelin.scheduler.Job.run(Job.java:176)
    at org.apache.zeppelin.scheduler.FIFOScheduler$1.run(FIFOScheduler.java:139)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    at java.util.concurrent.FutureTask.run(FutureTask.java:262)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:178)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:292)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.RuntimeException: Unable to instantiate org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient
    at org.apache.hadoop.hive.metastore.MetaStoreUtils.newInstance(MetaStoreUtils.java:1523)
    at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.<init>(RetryingMetaStoreClient.java:86)
    at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.getProxy(RetryingMetaStoreClient.java:132)
    at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.getProxy(RetryingMetaStoreClient.java:104)
    at org.apache.hadoop.hive.ql.metadata.Hive.createMetaStoreClient(Hive.java:3005)
    at org.apache.hadoop.hive.ql.metadata.Hive.getMSC(Hive.java:3024)
    at org.apache.hadoop.hive.ql.session.SessionState.start(SessionState.java:503)
    ... 58 more
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
    at org.apache.hadoop.hive.metastore.MetaStoreUtils.newInstance(MetaStoreUtils.java:1521)
    ... 64 more
Caused by: javax.jdo.JDOFatalInternalException: Unexpected exception caught.
NestedThrowables:
java.lang.reflect.InvocationTargetException
    at javax.jdo.JDOHelper.invokeGetPersistenceManagerFactoryOnImplementation(JDOHelper.java:1193)
    at javax.jdo.JDOHelper.getPersistenceManagerFactory(JDOHelper.java:808)
    at javax.jdo.JDOHelper.getPersistenceManagerFactory(JDOHelper.java:701)
    at org.apache.hadoop.hive.metastore.ObjectStore.getPMF(ObjectStore.java:365)
    at org.apache.hadoop.hive.metastore.ObjectStore.getPersistenceManager(ObjectStore.java:394)
    at org.apache.hadoop.hive.metastore.ObjectStore.initialize(ObjectStore.java:291)
    at org.apache.hadoop.hive.metastore.ObjectStore.setConf(ObjectStore.java:258)
    at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:76)
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:136)
    at org.apache.hadoop.hive.metastore.RawStoreProxy.<init>(RawStoreProxy.java:57)
    at org.apache.hadoop.hive.metastore.RawStoreProxy.getProxy(RawStoreProxy.java:66)
    at org.apache.hadoop.hive.metastore.HiveMetaStore$HMSHandler.newRawStore(HiveMetaStore.java:593)
    at org.apache.hadoop.hive.metastore.HiveMetaStore$HMSHandler.getMS(HiveMetaStore.java:571)
    at org.apache.hadoop.hive.metastore.HiveMetaStore$HMSHandler.createDefaultDB(HiveMetaStore.java:624)
    at org.apache.hadoop.hive.metastore.HiveMetaStore$HMSHandler.init(HiveMetaStore.java:461)
    at org.apache.hadoop.hive.metastore.RetryingHMSHandler.<init>(RetryingHMSHandler.java:66)
    at org.apache.hadoop.hive.metastore.RetryingHMSHandler.getProxy(RetryingHMSHandler.java:72)
    at org.apache.hadoop.hive.metastore.HiveMetaStore.newRetryingHMSHandler(HiveMetaStore.java:5762)
    at org.apache.hadoop.hive.metastore.HiveMetaStoreClient.<init>(HiveMetaStoreClient.java:199)
    at org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient.<init>(SessionHiveMetaStoreClient.java:74)
    ... 69 more
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at javax.jdo.JDOHelper$16.run(JDOHelper.java:1965)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.jdo.JDOHelper.invoke(JDOHelper.java:1960)
    at javax.jdo.JDOHelper.invokeGetPersistenceManagerFactoryOnImplementation(JDOHelper.java:1166)
    ... 88 more
Caused by: org.datanucleus.exceptions.NucleusException: Error creating the MetaDataManager for API "JDO" : 
    at org.datanucleus.NucleusContext.getMetaDataManager(NucleusContext.java:1001)
    at org.datanucleus.api.jdo.JDOPersistenceManagerFactory.initialiseMetaData(JDOPersistenceManagerFactory.java:703)
    at org.datanucleus.api.jdo.JDOPersistenceManagerFactory.<init>(JDOPersistenceManagerFactory.java:511)
    at org.datanucleus.api.jdo.JDOPersistenceManagerFactory.createPersistenceManagerFactory(JDOPersistenceManagerFactory.java:301)
    at org.datanucleus.api.jdo.JDOPersistenceManagerFactory.getPersistenceManagerFactory(JDOPersistenceManagerFactory.java:202)
    ... 96 more
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
    at org.datanucleus.plugin.NonManagedPluginRegistry.createExecutableExtension(NonManagedPluginRegistry.java:631)
    at org.datanucleus.plugin.PluginManager.createExecutableExtension(PluginManager.java:325)
    at org.datanucleus.NucleusContext.getMetaDataManager(NucleusContext.java:995)
    ... 100 more
Caused by: org.datanucleus.metadata.InvalidClassMetaDataException: Class "org.apache.hadoop.hive.metastore.model.MConstraint" has MetaData yet the class cant be found. Please check your CLASSPATH specifications.
    at org.datanucleus.metadata.AbstractClassMetaData.loadClass(AbstractClassMetaData.java:579)
    at org.datanucleus.metadata.ClassMetaData.populate(ClassMetaData.java:166)
    at org.datanucleus.metadata.MetaDataManager$1.run(MetaDataManager.java:2918)
    at java.security.AccessController.doPrivileged(Native Method)
    at org.datanucleus.metadata.MetaDataManager.populateAbstractClassMetaData(MetaDataManager.java:2912)
    at org.datanucleus.metadata.MetaDataManager.populateFileMetaData(MetaDataManager.java:2735)
    at org.datanucleus.api.jdo.metadata.JDOMetaDataManager.loadXMLMetaDataForClass(JDOMetaDataManager.java:786)
    at org.datanucleus.api.jdo.metadata.JDOMetaDataManager.getMetaDataForClassInternal(JDOMetaDataManager.java:383)
    at org.datanucleus.api.jdo.metadata.JDOMetaDataManager$MetaDataRegisterClassListener.registerClass(JDOMetaDataManager.java:207)
    at javax.jdo.spi.JDOImplHelper.addRegisterClassListener(JDOImplHelper.java:462)
    at org.datanucleus.api.jdo.metadata.JDOMetaDataManager.<init>(JDOMetaDataManager.java:194)
    ... 107 more
Caused by: org.datanucleus.exceptions.ClassNotResolvedException: Class "org.apache.hadoop.hive.metastore.model.MConstraint" was not found in the CLASSPATH. Please check your specification and your CLASSPATH.
    at org.datanucleus.ClassLoaderResolverImpl.classForName(ClassLoaderResolverImpl.java:216)
    at org.datanucleus.ClassLoaderResolverImpl.classForName(ClassLoaderResolverImpl.java:368)
    at org.datanucleus.metadata.AbstractClassMetaData.loadClass(AbstractClassMetaData.java:569)
    ... 117 more

지금의 결론

Zeppelin의 에러는 제대로 쫓으면 슛할 수 있을지도 모르지만, 거기까지 시간은 걸릴 것 같지 않기 때문에, 일단 다음의 HDP의 릴리스를 기다리는 것이 좋을까. LLAP을 GA로 해 Zeppelin에서 (가능하면 Spark2에서) 부를 수 있도록 해주는 곳까지 정돈되어 있으면 최고.

라고는 좋지만, 또 시간을 찾아내려고 하려고 한다.

EMR에 넣어 주면 더욱 기쁘다고 생각하지만, 이것에는 spark-llap가 본가 Spark에 넣어지거나 Hortonworks에서 spark package로서 릴리스 될지 하고 나서가 아니면 할 수 없기 때문에, 아직 아직 시간은 걸릴 것 같다 .