개발자로 살아남기

스칼라쉘에서 스파크로 작업을 작업을 하고 처리시간을 확인하고 싶을때는 SparkSession의 time() 함수를 이용합니다. spark.time(실행 함수) spark.time(sql("select * from tbl")) AWS EMR이의 스파크에는 해당 함수가 없어서 SparkSession에서 함수를 가져와서 이용하면 됩니다. 다른 작업에서도 처리 시간을 알고 싶을 때 이용하면 됩니다. scala> time(sql("use db"))Time taken: 12 ms res7: org.apache.spark.sql.DataFrame = [] 위와 같이 처리하면 Time taken으로 처리 시간을 확인할 수 있습니다. SparkSession 스칼라 소스코드: 바로가기

스칼라(scala) 개발을 연습하기 위해서 이클립스에 스칼라 개발 환경을 설정하고, 빌드하여 jar 파일을 만드는 것까지 설정하는 방법을 정리해 보겠다. 우선 이클립스를 설치한다. 저는 가장 최신버전인 4.7 Oxygen 버전을 설치하였다. 그리고 Help -> Eclipse Market Place 에서 scala를 검색하여 Scala IDE 4.7.x 를 설치한다. 이 플러그인을 설치하면 스칼라 라이브러리도 설치된다. 설치후 File -> New -> Scala Project 를 선택하여 스칼라 프로젝트를 생성한다. 그리고 Scala Library container에서 오른쪽 버튼을 클릭하여 properties 로 들어가서 스칼라 라이브러리를 2.11로 변경한다. 현재 스파크의 스칼라 버전이 2.11이..

스파크의 RDD, DataFrame, DataSet의 차이에 대해서 RDD, DataFrame, DataSet RDD Spark 1.0에서 소개 Java, Scala의 객체를 처리 하는 방식으로 처리 가능 transformation 함수들을 제공하고 각 함수의 결과를 RDD로 생성 action 함수가 호출되어야 실제 동작 rdd.filter(.age >21).map(.last).saveAsObjectFile("result.txt") DataFrame Spark 1.3에서 프로젝트 텅스텐의 일부로 소개 데이터를 스키마 형태로 추상화 하여 처리 Catalyst 옵티마이저에 의해 효율적으로 처리 df.filter("age > 21") df.filter(df.col("age").gt(21)) DataSet 스파..

RDD operationRDD를 제어하는 API operation은 크게 2개의 타입Transformation: RDD에서 새로운 RDD를 생성하는 함수Action: RDD에서 RDD가 아닌 타입의 data 로 변환하느 함수들 RDDMR의 단점복잡하고 Multi-Stage 한 처리가 느림interactive 하고 ad-hoc 한 쿼리 실행이 느림효율적인 데이터 공유 도구를 생성하기 위해 나옴MR은 중간 단계를 파일을 이용하기 때문에 IO 비용이 많이 들어감스파크는 하드디스크 대신 RAM을 이용하자 문제가 생기면 복구는 Lineage를 생성하여 새로 만들자 동작의 순서를 기록해 DAG로 표현한 것을 Lineage라고 함RAM을 Read-Only로 사용 Spark의 Operationtransformation..

스파크의 클러스터 타입은 4가지가 있다. standalone스파크에 포함된 클러스터 관리자 YARNHadoop2의 관리자 https://hadoop.apache.org/docs/current/hadoop-yarn/hadoop-yarn-site/YARN.htmlMesosHadoop MapReduce와 다른 응용 서비스를 실행할 수 있는 범용 클러스터 관리자 http://mesos.apache.org/Kubernetes컨테이너화된 응용프로그램의 배치, 실행을 관리하는 클러스터 관리자 https://kubernetes.io/ 그리고 이 클러스터에 대한 배포 모드(deploy mode)는 2가지가 있다. client실행을 호출한 곳에서 드라이버가 생성cluster클러스터 내부에서 드라이버가 생성 https://..

스파크 실행중 메모리 부족으로 다음과 같은 오류가 발생하는 경우가 있다. Diagnostics: Container [pid=1,containerID=container_000001] is running beyond physical memory limits. Current usage: 1.4 GB of 1.4 GB physical memory used; 3.1 GB of 13.8 GB virtual memory used. Killing container. 이런 경우 다음과 같이 드라이버와 executor의 메모리 설정을 해주면 해결할 수 있다. spark.yarn.am.memory=3072Mspark.driver.memory=3072Mspark.executor.memory=4736Mspark.executor..

spark 처리중 다음과 같은 오류가 발생하는 경우가 있다. 아마도 shuffle 처리중 메모리가 부족해서 발생하는 것으로 추정된다. 이럴때는 spark.sql.shuffle.partitions 설정을 추가하면 된다. 다음 설정을 추가하여 처리하였다. spark.sql.shuffle.partitions=300spark.default.parallelism=300 diagnostics: User class threw exception: org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 2.0 failed 4 times, most recent failure: Lost task 1.3 in stage 2.0 (TI..

pyspark에서 spakr 2.x 버전을 이용할 경우 utf-8 버전의 파일을 읽어서 처리할 경우 유니코드 오류가 발생한다. 이럴때 다음과 같이 처리하면 된다. ScalaContext를 이용하여 파일을 읽을 때 use_unicode 옵션을 이용하고, map 함수에서 라인단위로 읽을 때 인코딩을 처리해주면 된다.

- 2011년 트위터가 인수한 백타이프가 처음 시작한 프로젝트 - 트위터에서 오픈소스화 -> 아파치 프로젝트로 이동 - 클로저(Clojure)를 주언어로 사용 - 클로저에 기반을 두고 있지만 JVM에서 실행되며, 볼트는 다양한 언어로 개발이 가능 - 확장성이 크고, 내결함성이 강한, 스트림 프로세싱에 최적화된 이벤트 스트림 프로세싱 중심 - UC 버클리 산하 AMPLap에서 시작하여 아파치 프로젝트로 이동 - 범용 분산형 컴퓨팅 플랫폼 - 하둡을 기반으로 처리(불편하지만 하둡 없이도 가능함) - 스칼라, 파이썬을 이용한 인터랙티브 쉘 제공 - 확장성이 높은 실시간 분석과 BI 시스템을 위한 우수한 플랫폼 http://www.itworld.co.kr/news/91022 https://storm.apache..

아파치 스파크 실시간 분산형 컴퓨팅 프로젝트이다. 스트림 지향형 프로세스를 지원한다. 기존 하둡 클러스터에서 실행시킬 수 있다. 스칼라로 작성이 되어 있지만, 스칼라, 자바, 파이선 API를 지원한다. In-Memory 방식으로 처리를 하기 때문에 하둡에 비해 처리속도가 빠르다. 하둡에 비해 100배정도 빠르다고 한다. 예제 코드 # /spark/bin/pyspark 를 실행후 콘솔에서 입력 # README.md 파일은 hdfs 상에 업로드 되어 있어야 함 # hdfs://user/name/README.md 에 위치 # README.md 파일을 입력하여 라인 카운트 확인 lines = sc.textFile("README.md") print lines.count() # README.md 파일의 문장을 공백..