Apache Spark는 세 가지 유형의 API를 제공합니다

1. RDD
2. DataFrame
3. 데이터 세트

다음은 RDD, Dataframe 및 Dataset 간의 API 비교입니다.

RDD

스파크가 제공하는 주요 추상화는 RRD (Resilient Distributed Dataset)로, 클러스터 노드에서 분할되어 병렬로 작동 할 수있는 요소의 모음입니다.

RDD 특징 :-

• 분산 콜렉션 :
  RDD는 클러스터에서 병렬 분산 알고리즘을 사용하여 대용량 데이터 세트를 처리하고 생성하는 데 널리 채택되는 MapReduce 조작을 사용합니다. 사용자는 작업 분포 및 내결함성에 대해 걱정할 필요없이 일련의 고급 연산자를 사용하여 병렬 계산을 작성할 수 있습니다.

• 불변 : RDD는 파티션 된 레코드 콜렉션으로 구성됩니다. 파티션은 RDD의 기본 병렬 처리 단위이며 각 파티션은 기존 파티션의 일부 변환을 통해 변경 불가능하고 생성되는 데이터의 논리적 분할입니다.

• 내결함성 :
  RDD의 일부 파티션이 손실 된 경우 여러 노드에서 데이터 복제를 수행하지 않고 동일한 계산을 달성하기 위해 해당 파티션의 변환을 계보로 재생할 수 있습니다. 데이터 관리 및 복제에 많은 노력을 기울여 더 빠른 계산을 달성합니다.

• 지연 평가 : Spark의 모든 변환은 결과가 바로 계산되지 않기 때문에 지연됩니다. 대신, 일부 기본 데이터 세트에 적용된 변환 만 기억합니다. 변환은 조치가 드라이버 프로그램으로 결과를 리턴해야하는 경우에만 계산됩니다.

• 기능 변환 :
  RDD는 두 가지 유형의 작업, 즉 기존 데이터에서 새 데이터 집합을 만드는 변환과 데이터 집합에서 계산을 실행 한 후 드라이버 프로그램에 값을 반환하는 작업을 지원합니다.

• 데이터 처리 형식 :
  구조화되지 않은 데이터뿐만 아니라 구조화 된 데이터도 쉽고 효율적으로 처리 할 수 ​​있습니다.

• 프로그래밍 언어 지원 :
  RDD API는 Java, Scala, Python 및 R에서 사용할 수 있습니다.

RDD 제한 사항 :-

• 내장 된 최적화 엔진 없음 :
  구조화 된 데이터로 작업 할 때 RDD는 촉매 최적화 및 텅스텐 실행 엔진을 포함한 Spark의 고급 최적화 도구를 활용할 수 없습니다. 개발자는 속성에 따라 각 RDD를 최적화해야합니다.

• 구조화 된 데이터 처리 : 데이터 프레임
  및 데이터 세트와 달리 RDD는 수집 된 데이터의 스키마를 유추하지 않으며 사용자가이를 지정해야합니다.

데이터 프레임

Spark는 Spark 1.3 릴리스에서 데이터 프레임을 도입했습니다. 데이터 프레임은 RDD의 주요 과제를 극복합니다.

DataFrame은 명명 된 열로 구성된 분산 된 데이터 모음입니다. 관계형 데이터베이스 또는 R / Python 데이터 프레임의 테이블과 개념적으로 동일합니다. Spark는 Dataframe과 함께 고급 프로그래밍 기능을 활용하여 확장 가능한 쿼리 최적화 프로그램을 구축하는 촉매 최적화 도구도 도입했습니다.

데이터 프레임 특징 :-

• 행 개체의 분산 컬렉션 :
  DataFrame은 명명 된 열로 구성된 분산 데이터 컬렉션입니다. 개념적으로 관계형 데이터베이스의 테이블과 동일하지만 기본적으로 풍부한 최적화 기능이 있습니다.

• 데이터 처리 :
  정형 및 비정형 데이터 형식 (Avro, CSV, 탄력적 검색 및 Cassandra) 및 스토리지 시스템 (HDFS, HIVE 테이블, MySQL 등) 처리 이 모든 다양한 데이터 소스에서 읽고 쓸 수 있습니다.

• Catalyst Optimizer를 사용한 최적화 :
  SQL 쿼리와 DataFrame API를 모두 지원합니다. 데이터 프레임은 4 단계에서 촉매 트리 변환 프레임 워크를 사용합니다.

   1.Analyzing a logical plan to resolve references
   2.Logical plan optimization
   3.Physical planning
   4.Code generation to compile parts of the query to Java bytecode.
  

• Hive 호환성 :
  Spark SQL을 사용하면 기존 Hive웨어 하우스에서 수정되지 않은 Hive 쿼리를 실행할 수 있습니다. Hive 프런트 엔드 및 MetaStore를 재사용하고 기존 Hive 데이터, 쿼리 및 UDF와 완벽하게 호환됩니다.

• 텅스텐 :
  텅스텐은 메모리를 명시 적으로 관리하고 식 평가를 위해 바이트 코드를 동적으로 생성하는 물리적 실행 백엔드를 제공합니다.

• 프로그래밍 언어 지원 :
  Dataframe API는 Java, Scala, Python 및 R로 제공됩니다.

데이터 프레임 제한 :-

• 컴파일 타임 유형 안전성 :
  논의 된 바와 같이, Dataframe API는 구조를 모르는 경우 데이터 조작을 제한하는 컴파일 타임 안전성을 지원하지 않습니다. 다음 예제는 컴파일 시간 동안 작동합니다. 그러나이 코드를 실행할 때 런타임 예외가 발생합니다.

예:

case class Person(name : String , age : Int)
val dataframe = sqlContext.read.json("people.json")
dataframe.filter("salary > 10000").show
=> throws Exception : cannot resolve 'salary' given input age , name

여러 변환 및 집계 단계로 작업 할 때 특히 어려운 문제입니다.

• 도메인 객체 (손실 된 도메인 객체)에서 작동 할 수 없음 :
  도메인 객체를 데이터 프레임으로 변환 한 후에는 도메인 객체를 다시 생성 할 수 없습니다. 다음 예에서는 personRDD에서 personDF를 만든 후에는 원래 RDD of Person 클래스 (RDD [Person])를 복구하지 않습니다.

예:

case class Person(name : String , age : Int)
val personRDD = sc.makeRDD(Seq(Person("A",10),Person("B",20)))
val personDF = sqlContext.createDataframe(personRDD)
personDF.rdd // returns RDD[Row] , does not returns RDD[Person]

데이터 세트 API

Dataset API는 형식에 안전한 객체 지향 프로그래밍 인터페이스를 제공하는 DataFrames의 확장입니다. 관계형 스키마에 매핑되는 강력한 형식의 변경 불가능한 개체 컬렉션입니다.

데이터 세트의 핵심에서 API는 엔코더라고하는 새로운 개념으로, JVM 객체와 테이블 형식 간의 변환을 담당합니다. 테이블 형식 표현은 Spark 내부 텅스텐 이진 형식을 사용하여 저장되므로 직렬화 된 데이터에 대한 작업이 가능하고 메모리 사용률이 향상됩니다. Spark 1.6은 기본 유형 (예 : String, Integer, Long), Scala 케이스 클래스 및 Java Bean을 포함하여 다양한 유형에 대한 인코더 자동 생성을 지원합니다.

데이터 셋 특징 :-

• 
  RDD (기능 프로그래밍, 형식 안전), DataFrame (관계형 모델, 쿼리 최적화, 텅스텐 실행, 정렬 및 셔플 링) : RDD 및 데이터 프레임 모두를 제공합니다.

• 인코더 : 인코더를
  사용하면 모든 JVM 객체를 데이터 세트로 쉽게 변환 할 수 있으므로 사용자는 데이터 프레임과 달리 구조화 된 데이터와 구조화되지 않은 데이터를 모두 사용할 수 있습니다.

• 프로그래밍 언어 지원 :
  Datasets API는 현재 Scala 및 Java에서만 사용할 수 있습니다. Python 및 R은 현재 버전 1.6에서 지원되지 않습니다. Python 지원은 버전 2.0에 예정되어 있습니다.

• Type Safety :
  Datasets API는 데이터 프레임에서 사용할 수 없었던 컴파일 시간 안전을 제공합니다. 아래 예제에서 컴파일 람다 함수를 사용하여 도메인 객체에서 데이터 세트가 작동하는 방법을 확인할 수 있습니다.

예:

case class Person(name : String , age : Int)
val personRDD = sc.makeRDD(Seq(Person("A",10),Person("B",20)))
val personDF = sqlContext.createDataframe(personRDD)
val ds:Dataset[Person] = personDF.as[Person]
ds.filter(p => p.age > 25)
ds.filter(p => p.salary > 25)
 // error : value salary is not a member of person
ds.rdd // returns RDD[Person]

• 상호 운용성 : 데이터 세트를 사용하면 기존 RDD 및 데이터 프레임을 상용구 코드없이 데이터 세트로 쉽게 변환 할 수 있습니다.

데이터 세트 API 제한 :-

• String으로 타입 캐스팅이 필요합니다 :
  현재 데이터 셋에서 데이터를 쿼리하려면 클래스의 필드를 문자열로 지정해야합니다. 데이터를 쿼리 한 후에는 필요한 데이터 유형으로 열을 캐스트해야합니다. 반면에 데이터 세트에서 맵 작업을 사용하면 Catalyst 옵티마이 저가 사용되지 않습니다.

예:

ds.select(col("name").as[String], $"age".as[Int]).collect()

Python 및 R 지원 안 함 : 릴리스 1.6부터 데이터 세트는 Scala 및 Java 만 지원합니다. Python 2.0에 Spark 지원이 도입 될 예정입니다.

Datasets API는 기존 RDD 및 Dataframe API에 비해 형식 안전성 및 기능 프로그래밍이 향상되어 여러 가지 이점을 제공합니다.

답변

val rowsRdd: RDD[Row] = sc.parallelize(
  Seq(
    Row("first", 2.0, 7.0),
    Row("second", 3.5, 2.5),
    Row("third", 7.0, 5.9)
  )
)

val df = spark.createDataFrame(rowsRdd).toDF("id", "val1", "val2")

df.show()
+------+----+----+
|    id|val1|val2|
+------+----+----+
| first| 2.0| 7.0|
|second| 3.5| 2.5|
| third| 7.0| 5.9|
+------+----+----+

val rowsRdd: RDD[Row] = df.rdd() // DataFrame to RDD

val people = sqlContext.read.parquet("...")
val department = sqlContext.read.parquet("...")

people.filter("age > 30")
  .join(department, people("deptId") === department("id"))
  .groupBy(department("name"), "gender")
  .agg(avg(people("salary")), max(people("age")))

val people: People = val people = sqlContext.read.parquet("...").as[People]

val sampleRDD = sqlContext.jsonFile("hdfs://localhost:9000/jsondata.json")

val sample_DF = sampleRDD.toDF()