スパーク定義ガイド

第7章. アグリゲーション

この作品はAIを使って翻訳されている。ご意見、ご感想をお待ちしている：[e-mail]

集計とは、 何かをまとめて収集する行為であり、ビッグデータ分析の要である。集計では、キーまたはグループ化と、1つまたは複数の列をどのように変換するかを指定する集計関数を指定する。この関数は、複数の入力値が与えられた場合、グループごとに1つの結果を生成しなければならない。Sparkの集約機能は洗練されており、成熟している。一般化では、数値データをグループ化して要約するために集約を使用する。これは、合計、積、または単純なカウントかもしれない。また、Sparkでは、"複雑な型への集約 "で説明するように、あらゆる種類の値を配列、リスト、マップに集約することができる。

Sparkでは、どのような型の値でも扱えることに加え、以下のようなグループ化タイプを作成することもできる：

• 最も単純なグループ化は、select文の中で集計を行うことによって、DataFrame全体を要約することである。

• group by」では、1つ以上のキーと、値列を変換する1つ以上の集約関数を指定できる。

• ウィンドウ」では、1つ以上のキーと、値の列を変換する1つ以上の集約関数を指定することができる。ただし、関数に入力される行は、現在の行と何らかの関係がある。

• グループ化セット」は、複数の異なるレベルで集約するために使用できる。グループ化セットは、SQLではプリミティブとして、DataFramesではロールアップやキューブとして利用できる。

• ロールアップ」では、1つまたは複数のキーと、値の列を変換する1つまたは複数の集約関数を指定することができる。

• キューブ」では、1 つまたは複数のキーと、値列を変換する 1 つまたは複数の集約関数を指定でき、これらの関数は列のすべての組み合わせにわたって要約される。

それぞれの グループ化は、RelationalGroupedDataset 、その上で集約を指定する。

注

重要なことは、どの程度正確な答えが必要かということである。ビッグデータに対して計算を実行する場合、質問に対して正確な答えを得るにはかなりのコストがかかる。本書の中で、いくつかの近似関数について触れているが、これはSparkジョブのスピードと実行を向上させる良い機会である。

まずは購入したデータを読み取り、データのパーティション分割を大幅に減らし（小さなファイルにたくさん保存された少量のデータだとわかっているからだ）、結果をキャッシュして素早くアクセスできるようにしよう：

// in Scala
val df = spark.read.format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .load("/data/retail-data/all/*.csv")
  .coalesce(5)
df.cache()
df.createOrReplaceTempView("dfTable")

# in Python
df = spark.read.format("csv")\
  .option("header", "true")\
  .option("inferSchema", "true")\
  .load("/data/retail-data/all/*.csv")\
  .coalesce(5)
df.cache()
df ...