스파크의 RDD

• RDD : Spark 1.0부터 스파크에 도입된 가장 기초적인 데이터구조
• RDD의 세 가지 핵심 특징
  • 의존성(dependency) : 어떤 입력이 필요하고 생성되는 RDD가 어떻게 만들어지는지에 대한 정보
  • 파티션(partition)(지역성 정보 포함) : 작업을 나누어 이그제큐더들에 분산해 파티션별로 병렬 연산할 수 있는 능력 부여. 만약 파일을 읽는 경우 각 이그제큐터가 가까이 있는 데이터를 처리할 수 있는 이그제큐터에게 우선적으로 작업을 보냄.
  • 연산 함수(compute function) : 저장된 데이터를 Iterator[T] 형태로 만들어 줌.
• 문제점
  • 연산 함수나 연산식이 스파크에게 투명하지 않음 : 사용자가 연산 함수 안에서 어떤 작업을 하는지 스파크가 알지 못함. → 최적화할 방법이 없음.
  • Iterator[T] 데이터 타입이 RDD에서 불투명함. 단지 파이썬 기본 객체로 인식. → 데이터 압축 테크닉을 적용하지 못하고, 데이터를 바이트 뭉치로 직렬화하여 사용하는 방법만 존재.

스파크의 구조 확립

• 스파크 2.X부터 스파크 구조 확립을 위한 핵심 개념들을 도입함.

• 저수준의 RDD API 패턴과 고수준 DSL과 데이터 프레임 API를 사용한 예시 비교

  • 저수준의 RDD API 패턴 이용

  sc = spark.sparkContext
  # (name, age) 형태의 튜플로 된 RDD 생성
  dataRDD = sc.parallelize([("Brooke", 20), ("Denny", 31), ("Jules", 30), ("TD", 35), ("Brooke", 25)])
  # 집계와 평균을 위한 람다 표현식, map, reduceByKey transformation
  ageRDD = (dataRDD
            .map(lambda x: (x[0], (x[1], 1)))
            .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))
            .map(lambda x: (x[0], x[1][0] / x[1][1]))
            )
  print(ageRDD.collect())
  

  

  • 고수준 DSL, 데이터 프레임 API 사용

  data_df = spark.createDataFrame([("Brooke", 20), ("Denny", 31), ("Jules", 30), ("TD", 35), ("Brooke", 25)], ['name', 'age'])
  avg_df = data_df.groupBy('name').agg(avg('age'))
  avg_df.show()
  

  

  ⇒ 간단하고, 직관적인 코드 구현이 가능하다.

데이터 프레임 API

• 지원하는 프로그래밍 언어와 맞게 스파크는 기본적인 내부 데이터 타입을 지원함.

• 스키마(schema) : 데이터 프레임을 위해 칼럼 이름과 연관된 데이터 타입을 정의한 것.

  • 일반적으로 외부 데이터 소스에서 구조화된 데이터를 읽어 들일 때 쓰임.

• 데이터를 읽기 전 스키마를 정의하는 것의 장점

  • 스파크가 데이터 타입을 추측해야 하는 책임을 덜어 줌.
  • 스파크가 스키마를 확정하기 위해 파일의 많은 부분을 읽어 들이려고 잡을 만드는 것을 방지.
  • 데이터가 스키마와 맞지 않는 경우, 초기에 문제를 발견할 수 있음.

• 스키마를 정의하는 두 가지 방식

  • 프로그래밍 스타일

  # 프로그래밍 스타일
  from pyspark.sql.types import *
  schema = StructType([StructField("author", StructType(), False),
                       StructField("title", StringType(), False),
                       StructField("pages", IntegerType(), False)])
  

  • DDL(data definition language) - 더 간단함.

  # DDL 사용
  schema = "author STRING, title STRING, pages INT"
  

• 칼럼과 표현식

  • 데이터 프레임에서 정해진 칼럼들은 pandas나 R에서의 데이터 프레임이나 RDBMS 테이블의 칼럼과 유사하게 어떤 특정한 타입의 필드를 나타내는 개념임.
  • 칼럼을 나열하거나, 관계형 표현, 계산식 형태의 표현식으로 값들에 대한 연산을 수행할 수 있다.

• Example 3-6 : 데이터를 생성하고 데이터 프레임으로 정의하는 예제 - 링크

• 로우(row)

  • 스파크에서 하나의 행은 일반적으로 하나 이상의 칼럼을 가진 로우 객체로 표현됨.
  • Row는 스파크의 객체이고 순서가 있는 필드 집합 객체이므로 스파크의 지원 언어들에서 각 필드를 0부터 시작하는 인덱스로 접근함.

  from pyspark.sql import Row
  blog_row = Row(6, "Reynold", "Xin", "<https://tinyurl.6>", 255568, "3/2/2015", ["twitter", "LinkedIn"])
  blog_row[1]
  

  

  # 데이터 프레임으로 만들어서 사용
  rows = [Row("Matei Zaharia", "CA"),Row("Reynold Xin", "CA")]
  authors_df = spark.createDataFrame(rows, ["Authors", "State"])
  authors_df.show()
  

  

⭐️자주 쓰이는 데이터 프레임 작업들

DataFrameReader와 DataFrameWriter