Spark 애플리케이션은 클러스터의 드라이버 프로그램과 익스큐터 그룹으로 구성됩니다. Driver는 Spark 애플리케이션의 기본 프로그램을 실행하고 작업 실행을 조정하는 SparkContext를 생성하는 프로세스입니다. executor는 드라이버 프로세스가 할당 한 task을 실행하는 클러스터의 worker 노드에서 실행되는 프로세스입니다.

클러스터 관리자(예: Mesos 또는 YARN)는 Spark 애플리케이션에 대한 물리적 리소스 할당을 담당합니다.

spark 진입점

모든 Spark 애플리케이션에는 데이터 소스와 통신하고 데이터 읽기 및 쓰기와 같은 특정 작업을 수행할 수 있는 진입점이 필요합니다. Spark 1.x 에서는 SparkContext , SQLContext 및 HiveContext 의 세 가지 진입점이 도입되었습니다 . Spark 2.x 이후 세 가지 컨텍스트에서 사용할 수 있는 모든 기능을 기본적으로 결합한 SparkSession 이라는 새로운 진입점이 도입되었습니다.

Spark의 초기 릴리스에서는 각각 다른 목적을 가진 이 세 가지 진입점을 사용할 수 있었습니다.

SparkContext

SparkContext는 작업을 조정하고 실행하기 위해 클러스터 및 리소스 관리자와 통신을 설정하기 위해 드라이버 프로세스에서 사용됩니다. SparkContext는 다른 두 컨텍스트, 즉 SQLContext 및 HiveContext에 대한 액세스도 가능하게 합니다 SparkContext를 생성하려면 먼저 아래와 같이 Spark 구성( SparkConf)을 생성해야 합니다.

// Scala import org.apache.spark.{SparkContext, SparkConf} val sparkConf = new SparkConf()
.setAppName("app")
.setMaster("yarn") val sc = new SparkContext(sparkConf)

spark-shell을 사용하는 경우 sc 라는 변수를 통해 SparkContext를 이미 사용할 수 있습니다 .

SQLContext

SQLContext는 구조화된 데이터 처리를 위한 Spark 모듈인 SparkSQL 의 진입점 입니다. SQLContext가 초기화되면 사용자는 이를 사용하여 데이터 세트 및 데이터 프레임에 대해 다양한 "sql-like" 작업을 수행할 수 있습니다.

SQLContext를 생성하려면 먼저 아래와 같이 SparkContext를 인스턴스화해야 합니다.

// Scala import org.apache.spark.{SparkContext, SparkConf} import org.apache.spark.sql.SQLContext val sparkConf = new SparkConf()
.setAppName("app")
.setMaster("yarn") val sc = new SparkContext(sparkConf) val sqlContext = new SQLContext(sc)

HiveContext

Spark 응용 프로그램이 Hive와 통신해야 하고 Spark < 2.0을 사용하는 경우 HiveContext 가 필요할 것입니다. Spark 1.5+의 경우 HiveContext는 window functions를 지원합니다.

// Scala import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.sql.hive.HiveContext val sparkConf = new SparkConf()
.setAppName("app")
.setMaster("yarn") val sc = new SparkContext(sparkConf) val hiveContext = new HiveContext(sc) hiveContext.sql("select * from tableName limit 0")

a) Hive 지원도 제공하는 SparkSession이 도입되었습니다.

b) Native window functions이 릴리스되어 기본적으로 Hive UDAF를 기본 Spark SQL UDAF로 대체했습니다.

SparkSession

Spark 2.0은 본질적으로 SQLContext와 HiveContext를 모두 대체하는 SparkSession 이라는 새로운 진입점을 도입했습니다. 또한 개발자에게 SparkContext에 대한 즉각적인 액세스를 제공합니다. Hive를 지원하는 SparkSession을 생성하려면 다음 작업을 수행하기만 하면 됩니다.

// Scala import org.apache.spark.sql.SparkSession val sparkSession = SparkSession
.builder()
.appName("myApp")
.enableHiveSupport()
.getOrCreate() // Two ways you can access spark context from spark session val spark_context = sparkSession._sc val spark_context = sparkSession.sparkContext

개발을 하면서 hivecontext를 사용하지만 둘의 명확한 차이가 매번 기억나지 않아 다시 정리해 본다.

(둘이 명확하게 다르구나! 하고 느낀건 hive의 UDF를 spark sql에서 호출하지 못했을때 였나? 너무 당연한 결과지만...ㅎㅎ)

sc = pyspark.SparkContext(conf=conf).getOrCreate() sqlContext = HiveContext(sc) sqlContext.sql ('select * from table')

spark.sql('select * from table')

둘의 가장 큰 차이점은 hive의 기능을 사용할건지 말건지에 있는것 같다. 기본적으로 hive context는 sparkcontext보다 상위 개념이다

• hiveContext는 hive UDF사용 가능, hive 테이블의 데이터를 읽거나 윈도우 함수를 사용할 수 있다

• HiveContext를 사용하면 기존 Hive 설정이 필요하지 않는다.

https://medium.com/@an_chee/why-using-mixed-case-field-names-in-hive-spark-sql-is-a-bad-idea-95da8b6ec1e0

https://hellomuzi.tistory.com/56

https://www.codenong.com/cs106837148/

스키마 추가 했을 때 문제 https://stackoverflow.com/questions/60040363/spark-sql-does-not-pick-up-changes-in-hive-table-schema

--num-executors 39 --executor-memory 20g --executor-cores 5 \

→ 컨테이너 안에 excutor 생성 되므로 yarn UI에서는 컨테이너 40개와 (익스큐터 +1로 생김) vcore는 동일하게 40개로 생김

vcore의 경우 yarn-site.xml에 min = 1이기 때문에 컨테이너 수만큼 일정하게 생성

설정한 spark의 excutor-core의 경우 익스큐터 안에서 5개의 task로 도는거기 때문에 spark ui에서 확인하면 core가 195개로 돌아가고 있음을 확인 할 수 있다

INSERT가 되지 않아 디버깅을 하다보니 hive에서는 select가 되지만 spark에서는 spark sql로 select가 되지 않고,

spark.write.parquet로 hdfs파일을 읽으면 데이터가 보이는 현상

에러 WARN HiveMetastoreCatalog: Unable to infer schema for table 테이블명 from file format Parquet (inference mode: INFER_AND_SAVE). Using metastore schema.

이게 힌트가 아닐까 한다..

이는 CREATE 를 SPARK로 하면 이런 현상이 없어짐을 알 수 있는데, 생성된 DDL을 살펴보면 이유를 알 수 있었다. desc formatted 테이블 명

SPARK로 시작하는 부분이 모두 존재 해야 정상적으로 데이터를 읽을 수 있다

예를 들어 spark에서 wordcount app을 실행시킨다고 하자. 이 app이 1.5gb의 memory와 2개의 코어가 필요하다고 yarn에게 알려주면, yarn은 2gb의 memory와 2개의 코어를 갖는 container 를 만들어서 제공한다.

그러면 app은 해당 container를 받고 그 위에서 일을 한다. (일이 마무리 된 후 container(자원)는 yarn 에게 돌려줌)

yarn이 container로 만들 수 있는 total memory는 (yarn.nodemanager.resource.memory-mb) * (num of node managers)이다.

아래의 yarn.nodemanager.resource.memory-mb값이랑 노드 매니저의 개수를 곱하면 total memory가 나온다.

yarn 관련 모니터링을 위한 webUI 가 있다. [ip address]:8088 로 접속하여 볼 수 있다.

여기 그림이 배경 지식을 이해하기에 굉장히 잘 나와있어서 참고한다. https://docs.cloudera.com/documentation/enterprise/5-5-x/topics/cdh_ig_yarn_tuning.html

yarn을 설정할 때는 yarn-site.xml을 살펴보면 된다. yarn-site.xml내부 옵션 값들을 직접 지정해줄 수 있는데 설명은 다음과 같다. 참고로 master node뿐 아니라 worker node의 옵션들도 바꿀 수 있다.

memory관련 옵션

< yarn.nodemanager.resource.memory-mb > 이 옵션은 각 노드 매니저가 컨테이너 할당에 사용할 메모리 크기를 나타낸다. 서버마다 yarn 에서 관리하고 있는 노드 매니저가 하나씩 있는데(서버 당 하나씩 총 20개),

이 노드 매니저가 container 를 만들 때 사용할 수 있는 메모리 크기를 말한다. 예를 들어, 위의 값이 20gb 라면, 1gb 짜리 container 를 20개 만들 수 있고, 5gb 짜리 container 4개를 만들 수 있다.

우리의 가정에서는 하나의 서버에 128gb 메모리가 있으므로 yarn.nodemanager.resource.memory-mb 값을 128gb 로 잡아두면 하나의 서버가 갖는 모든 메모리를 yarn 이 사용하게 한다는 의미가 된다.

하지만 그렇게 하지 않는데, 이유는 서버가 yarn말고 돌려야 할 다른 소프트웨어들이 많기 때문이다. 예를 들면 서버 os. 따라서 하나의 서버의 전체 메모리 크기에서 적정 양만큼 제외한 값을 container 만드는 데 사용할 수 있도록 yarn.nodemanager.resource.memory-mb 값을 지정한다. 예를 들어 서버 하나의 크기가 128gb 니까, 8gb 는 os가 돌아갈 메모리로 남겨두고 남은 120gb 를 yarn.nodemanager.resource.memory-mb 값으로 지정하는 것이다. yarn.nodemanager.resource.memory-mb의 default 값은 8gb이다.

< yarn.scheduler.maximum-allocation-mb > 이 옵션은 container 에 할당할 수 있는 최대 memory 크기를 의미한다. 기본값은 8gb 이다. 기본값 8gb 적용되었다는 상황을 가정해보자. 예를 들어 spark에서 어떤 app을 구동시키는 데 6gb memory가 필요하다면, 8gb 보다 작기 때문에 6gb 만큼 container 를 만들어서 app에 제공할 수 있다. 만약 app 을 구동시키는 데 10gb memory 가 필요하다면, 8gb 보다 크기 때문에 container 를 만들어서 제공할 수 없어 에러를 내보낸다.

< yarn.scheduler.minimum-allocation-mb > 이 옵션은 container에 할당하는 memory의 단위를 의미한다. 기본값은 1gb이다. 기본값 1gb가 적용되었다는 상황을 가정해보자. 예를 들어 spark에서 어떤 app을 구동시키는 데 800mb memory가 필요하다면, 1gb container 를 만들어서 app에 제공한다. 왜냐하면 1gb가 memory를 제공하는 단위이기 때문이다.

만약 app을 구동시키는 데 1.9gb memory가 필요하다면, 2gb container 를 만들어서 app에 제공한다. 만약 app 을 구동시키는 데 2.1gb memory가 필요하다면, 3gb container 를 만들어서 app에 제공한다.

cpu(코어) 관련 옵션

< yarn.nodemanager.resource.cpu-vcores > 이 옵션은 각 노드 매니저가 컨테이너 할당에 사용할 cpu코어 개수를 의미한다. 노드 매니저가 container를 만들 때 사용할 수 있는 코어 개수를 말한다. 예를 들어 코어 개수가 20개라면, 1코어짜리 container 20개를 만들 수 있고, 2코어짜리 container 10개를 만들 수 있겠다. 기본 값은 8이다.

< yarn.scheduler.maximum-allocation-vcores > 이 옵션은 container 에 할당할 수 있는 최대 viertual cpu 코어 수를 의미한다. maximum 코어 수가 10이라는 상황을 가정해보자. 예를 들어 spark 에서 어떤 app을 구동시키는 데 5개 코어가 필요하다면, 10개 보다 작기 때문에 5개 코어 만큼 container를 만들어서 app에 제공할 수 있다. 만약 app을 구동시키는 데 12개 코어가 필요하다면, 10개 보다 크기 때문에 container를 만들어서 제공할 수 없어 에러를 내보낸다.

< yarn.scheduler.minimum-allocation-vcores > 이 옵션은 container에 할당하는 virtual cpu코어의 단위를 의미한다. minimum 코어 수가 3이라는 상황을 가정해보자. 예를 들어 spark 에서 어떤 app을 구동시키는 데 1개의 코어가 필요하다면, 3개 코어 만큼의 container를 만들어서 app 에 제공한다. 왜냐하면 3개가 cpu를 제공하는 단위이기 때문이다. 만약 app 을 구동시키는 데 4개 코어가 필요하다면, 6개 코어 만큼의 container 를 만들어서 app 에 제공한다.

https://hadoop.apache.org/docs/r3.1.1/hadoop-yarn/hadoop-yarn-common/yarn-default.xml

참고 https://m.blog.naver.com/gyrbsdl18/220594197752

https://docs.cloudera.com/documentation/enterprise/5-5-x/topics/cdh_ig_yarn_tuning.html

컨테이너 개수 계산 : https://blrunner.tistory.com/103

https://wikidocs.net/27332

https://stackoverflow.com/questions/29940711/apache-spark-setting-executor-instances-does-not-change-the-executors

https://community.cloudera.com/t5/Support-Questions/from-where-we-get-the-Yarn-memory-value-from-the-dashboard/td-p/228709

어떤형태의 데이터 크기와 구성에도 사용 가능 가장 자원을 많이 사용하고 느린 조인 방식

맵조인 맵 조인은 두 개의 테이블을 조인할 때 하나의 테이블이 메모리에 로드 되어 처리됩니다. 따라서 하나의 테이블이 메모리에 올라갈 수 있을 정도로 작을 때 맵조인을 적용할 수 있습니다. hive.auto.convert.join이 true일 때 적용되고 hive.auto.convert.join.noconditionaltask.size이 메모리에 올릴 테이블의 기본 사이즈로 10MB으로 설정되어 있습니다.

셔플 조인에 비하여 빠른 속도로 처리할 수 있음 테이블이 메모리에 올라갈 수 있는 크기여야 함 -- 맵조인 적용을 위한 설정. 기본 10MB로 설정 set hive.auto.convert.join=true; set hive.auto.convert.join.noconditionaltask.size=10000000;

SMB 조인 SMB 조인은 조인 테이블이 버켓팅 되어 있을 때 사용할 수 있습니다. 버켓팅된 키의 정보를 이용하여 빠르게 조인을 처리할 수 있습니다. 어떤 크기의 테이블에서도 가장 빠른 속도로 조인을 처리할 수 있습니다.

테이블이 버켓팅이 되어 있어야 함

-- SMB 조인을 위한 설정 set hive.auto.convert.sortmerge.join=true; set hive.optimize.bucketmapjoin=true; set hive.optimize.bucketmapjoin.sortedmerge=true;

https://wikidocs.net/83198

하이브에서 조인을 처리할때, 마지막 테이블을 가장 크다고 가정한다. 다른 테이블을 버퍼링하려고 시도하고, 각 레코드에 대해서 조인을 수행하면서 마지막 테이블을 흘려보낸다. 그러므로 조인쿼리를 구성할때, 가장 큰 테이블을 가장 마지막에 위치하도록 해야한다. 가장 큰 테이블을 쿼리의 마지막에 위치하지 못하였더라도 힌트를 사용하면 마지막에 처리할 수 있다. 아래 스크립트는 힌트를 사용하여 from의 첫 테이블을 마지막에 처리할 수 있도록 한다.

left outer join보다는 inner join을 쓴다.

• outer join보다는 inner join이 성능이 더 좋단다.

count(distinct ..)는 inner view의 group by로 푼다

• http://stackoverflow.com/questions/19311193/why-is-countdistinct-slower-than-group-by-in-hive
• count(distinct ..)가 있는 쿼리는 하나의 reducer에서 distinct 처리까지 수행한다!!
• group by로 풀어줘야 명시적으로 여러 reducer에 흩어서 처리한다.

불필요한 null, '' 등을 미리 제거한다!

• group by용 키의 값이 null, '' 등인 경우는 대부분쓸데없다.
• 그런데 ''인 값이 매핑되는 경우가 무수히 많을 경우 reducer 하나가 이 값을 처리해야 해서 느려질 수 있다.
• 꼭 필요한 경우가 아니면 group by 속성의 null, '' 등은 미리 where 조건에서 filter out 시키자.

ODS 운영계 시스템의 데이터는 트랜잭션을 중심으로 설계되어 있기 때문에, 운영계 시스템 데이터를 대상으로 데이터 분석을 진행할 경우 효용성이 매우 떨어집니다. 그래서 데이터 분석을 위한 DW(Data Warehouse)를 설계하고, 데이터를 보관하게 됩니다. ODS는 EDW로 데이터를 저장하기 전에, 임시로 운영계 데이터를 보관하는 장소이며, 운영계 시스템의 이력성 데이터를 보관을 하게 됩니다. 참고로 ODS가 별도로 존재할 경우에는 운영계 시스템의 복사본으로 활용되며, DW의 일부일 경우에는 임시 저장소로 사용됩니다.

EDW의 정의 ODS를 거쳐 운영 데이터베이스(Operational Database) 및 외부 데이터 Source로부터 필요 데이터를 추출하여, 경영분석/의사결정의 지원을 위해 최적화된 구조로 변환된 데이터베이스.

EDW의 특징

1. 주제 중심적 (Subject oriented)
2. 통합적 Integrated)
3. 계열적 (Time Variant Historical)
4. 장기 지속적 (Non-Volatile)

DM (Data Mart) OLAP분석의 편의성을 높이고자, EDW의 데이터를 주제별, 업무별로 요약하여 구성한 데이터 저장소입니다. 일반적으로 각 부서별로 다양한 예측과 분석을 목표로 만들어집니다.

OLAP 고객의 업무를 분석하고 다차원 모델링을 활용하여 Data Mart를 구축하고, OLAP (On Line Analysis Process)를 구현함으로써, 시각화 도구, 데이터 마이닝 시스템이 이 분류에 포함됩니다.

ETL(Extraction, Transform, Load) ETL은 데이터를 추출, 가공, 적재하는 프로세스입니다. 운영계 시스템과 ODS, ODS와 DW, DW와 DM간에 ETL이 진행됩니다.

LLAP은 asynchronous spindle-aware IO, prefetching and caching of column chunks, multi-threaded JIT-friendly operator pipelines 를 제공하여 hive의 성능을 향상 시켰다.

https://blog.cloudera.com/choosing-the-right-data-warehouse-sql-engine-apache-hive-llap-vs-apache-impala/

LLAP를 사용하지만 속도가 저하된 경우는?

LLAP은 조인과 집계가 관련 된 쿼리에 최적화 되어 있습니다. 다음과 같은 쿼리는 대화형 Hive 클러스터에서 제대로 작동 하지 않습니다.

select * from table where column = "columnvalue"

Hive LLAP에서 point 쿼리 성능을 향상 시키려면 다음 구성을 설정 합니다.

hive.llap.io.enabled=false; (disable LLAP IO)
hive.optimize.index.filter=false; (disable ORC row index)
hive.exec.orc.split.strategy=BI; (to avoid recombining splits)

또한 다음과 같은 구성 변경으로 사용 하 여 성능을 향상 시킬 수 있습니다.

hive.fetch.task.conversion=none