Hee'World

- Spark Streaming은 실시간 데이터 스트림을 확장 가능하고 많은 처리량의 내결함성 스트림 처리를 지원하는 Spark의 핵심 API - 카프카, Flume, S3, HDFS, TCP 소켓과 같은 다양한 소스로부터 수집 할 수 있음 - 처리된 데이터를 파일 시스템, 데이터베이스 및 라이브 대시 보드로 - 실제로 Spark의 기계 학습 및 그래프 처리 알고리즘을 데이터 스트림에 적용 할 수 있음 Structured Streaming - 실제 Spark 2.x로 와서는 Structuured Streaming을 권장하고 있으며, 스트림 데이터를 Spark DataFrame의 하나의 행으로 계속 추가해서 처리한다고 보면 된다. - 이벤트 시간이라는 데이터 기록된 시간 필드를 기준으로 처리 - 워터마크라..

- Spark 이전, SQL on Hadoop으로 Hive가 사실상 표준 - DataFrame을 createOrReplaceTempView로 등록하여 SQL 사용 가능 - Grobal TempView • Spark Session 전역에서 사용 가능하도록 선언 • createOrReplaceTempView는 현재 SparkSession에서만 사용 가능 Spark SQL¶ Spark DataFrame을 Database Table처럼 사용 In [1]: import pandas as pd Pandas 데이터프레임 생성¶ In [5]: pandf = pd.read_csv("data/Uber-Jan-Feb-FOIL.csv", header=0) In [6]: pandf.head() Out[6]: dispatchin..

Titanic DataFrame 생성¶ In [1]: import pandas as pd In [15]: data1 = {'PassengerId':{0:1, 1:2, 2:3, 3:4, 4:5}, 'Name' : {0:'Owen', 1:'Florence', 2:'Laina', 3:'Lily', 4:"William"}, 'sex' : {0: 'male', 1: 'female', 2:'female', 3:'female', 4:'male'}, 'Survived': {0:0, 1:1, 2:1, 3:1, 4:0} } data2 = ..

Spark DataFrame¶ - select¶ In [3]: df = spark.read.json("data/2015-summary.json") In [6]: df.printSchema() root |-- DEST_COUNTRY_NAME: string (nullable = true) |-- ORIGIN_COUNTRY_NAME: string (nullable = true) |-- count: long (nullable = true) In [8]: df.select("DEST_COUNTRY_NAME").show() +--------------------+ | DEST_COUNTRY_NAME| +--------------------+ | United States| | United States| | Unite..

Spark에서 Row와 Column의 형태로 RDD를 표현하여 처리 할 수 있음 타입 - Python의 Pandas 패키지의 DataFrame과 R의 DataFrame과 동일한 개념 - Spark 2.x에서 Catalyst Optimizer의 도입으로 인해 Spark에서 지원하는 프로그래밍 타입 별 처리 성능이 동일하게 향상되었음 Spark DataFrame¶ spark.read.csv() spark.read.json() spark.read.format("csv") spark.read.format("json") file:// hdfs:// hbase:// s3:// In [1]: stock = spark.read.csv("data/appl_stock.csv", inferSchema=True, header..

대화형 콘솔 - Spark는 기본적으로 Scala, Python, SQL, R 등 대화형 콘솔 명령 프롬프트를 제공하고 있다. sc 객체 - SparkContext라는 Spark에서 기존에 사용하였던 객체 sc SparkContext Spark UI Version v2.4.4 Master local[*] AppName PySparkShell sc 객체를 이용한 RDD 생성 rdd = sc.parallelize([1, 2, 3]) rdd.collect() [1, 2, 3] type(rdd) pyspark.rdd.RDD list_set = [('cat', 'dog'),(1, 2, 3)] rdd = sc.parallelize(list_set) rdd.collect() [('cat', 'dog'), (1, 2,..

RDD는 스파크에서 가장 중요하고 핵심적인 기본 데이터 타입이라고 볼 수 있다. 스파크에서 처리되는 모든 데이터는 RDD를 기본으로 처리되고 실행된다. RDD는 Resilient Distributed Dataset라고 불리며, 불변(변하지않는)하고 분산되는 데이터집합이라고 볼 수 있다. 스파크에서 각각의 데이터는 클러스터 메모리에 분산되어 Partition 단위로 분산 저장된다. 또한, 리니지(Lineage)라는 RDD 생성 단계를 기록하여 연산 처리 중, 노드의 장애 또는 실패가 발생 시 데이터를 재구성하여 다시 연산 할 수 있도록 한다. RDD 연산에서는 Action연산과 Transformation연산이라는 2가지 연산이 존재한다. 먼저 Transformation 연산은 RDD의 불변성과도 연관이 있..

아파치 스파크(Apache Spark)는 아래 공식 홈페이지에 소개된 것처럼 대용량 데이터를 처리하기 위한 통합 분산 엔진이다. 기존 하둡(Hadoop)이라는 플랫폼을 대체 했다기 보다는 완벽히 조화를 이루었다고 보는게 맞을거 같다. 아파치 스파크는 2009년 U.C. 버클리의 AMP랩(AMPLab)에서 시작되었으며, 지금은 빅데이터 분야에서 가장 중요한 빅데이터 분산 처리 프레임워크이다. 또한, 아파치 스파크는 다양한 데이터 소스, 개발환경, 작업등을 통합 분석 할 수 있는 엔진으로 개발되고 발전하고 있으며, 최근에는 딥러닝(Deeplearning)과 파이썬(Python)의 Pandas패키지등 인공지능과 데이터분석에도 적극적으로 지원하고 있다. 기존 Disk 기반으로 처리되는 MapReduce의 처리 ..