배경

Dataset API 등장

Let's give it a try

Spark 1.6.0 다운로드

Spark 소스에 동봉된 파일을 사용하여 비교해 보기

// 各自のパスを合わせて設定
scala> val peopleFile = "/usr/local/spark-1.6.0/examples/src/main/resources/people.json"

// JSON → DataFrame
scala> val df = sqlContext.read.json(peopleFile)
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

// 中身確認
scala> df.printSchema
root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)

scala> df.show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

// Datasetを使うため、case class 定義
scala> case class Person(age: Long, name: String)

// DataFrameからDatasetに変換
scala> val ds = df.as[Person]
ds: org.apache.spark.sql.Dataset[Person] = [age: bigint, name: string]

// DataFrameでの20歳以上の人を取得
// DataFrameは行列計算に特化したフレームワークなので、カラム名を指定する必要がある
scala> df.where($"age" >= 20).show
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+

// Datasetでの20歳以上の人を取得
// DatasetはDataFrameのRowをJVMオブジェクト（この例ではPerson）として扱えるため、UDFが適用が簡単
scala> ds.filter(_.age >= 20).show
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+

// Dataset →　DataFrame
scala> val df2 = ds.toDF
df2: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

// Dataset →　RDD
scala> val rdd = ds.rdd
rdd: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[121] at rdd at <console>:33

// 少し複雑な処理（年代別人数集計）
scala> import org.apache.spark.sql.types._

// DataFrameの場合
scala> :paste
df.where($"age" > 0)
  .groupBy((($"age" / 10) cast IntegerType) * 10 as "decade")
  .agg(count($"name"))
  .orderBy($"decade")
  .show

+------+-----------+
|decade|count(name)|
+------+-----------+
|    10|          1|
|    30|          1|
+------+-----------+

// Datasetの場合
scala> :paste
ds.filter(_.age > 0)
  .groupBy(p => (p.age / 10) * 10)
  .agg(count("name"))
   // orderByがないようなので、DFへ変換（これはやや不便）
  .toDF().withColumnRenamed("value", "decade").orderBy("decade") 
  .show

+------+-----------+
|decade|count(name)|
+------+-----------+
|    10|          1|
|    30|          1|
+------+-----------+

다른 예는 databricks 사이트를 참조하십시오.

Dataset API 이외의 Spark1.6.0에 대해서는 이쪽

요약