Apache Sparkとは何か

使い始めて3年くらい経ちますが、改めて振り返ってみます。

こちらの記事を書いたりしていますが復習も大事なわけで。

https://qiita.com/taka_yayoi/items/5415db284b96c7c60d44

2024/4/12に翔泳社より**Apache Spark徹底入門**を出版します！

<a href="https://www.amazon.co.jp/dp/4798182281/"><img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/87910d7c-8a63-c198-8e82-216ea7a8ee07.jpeg" width=300></a>


その他のDatabricksコアコンポーネントの記事はこちらです。

https://qiita.com/taka_yayoi/items/799b0320e4d8ae2e4234

https://qiita.com/taka_yayoi/items/07c9396809edbf2699b6

# Apache Sparkプロジェクトの歴史

![](https://upload.wikimedia.org/wikipedia/commons/thumb/f/f3/Apache_Spark_logo.svg/320px-Apache_Spark_logo.svg.png)

SparkはDatabricksの創始者たちがUC Berkeleyにいるときに誕生しました。Sparkプロジェクトは2009年にスタートし、2010年にオープンソース化され、2013年にApacheにコードが寄贈されApache Sparkになりました。Apache Sparkのコードの75%以上がDatabricksの従業員の手によって書かれており、他の企業に比べて10倍以上の貢献をし続けています。Apache Sparkは、多数のマシンにまたがって並列でコードを実行するための、洗練された分散処理フレームワークです。概要とインタフェースはシンプルですが、クラスターの管理とプロダクションレベルの安定性を確実にすることはそれほどシンプルではありません。Databricksにおいては、Apache Sparkをホストするソリューションとして提供することで、ビッグデータをシンプルなものにします。

# Apache Sparkとは

https://www.oreilly.com/library/view/learning-spark-2nd/9781492050032/ch01.html

[Apache Spark](https://spark.apache.org/)は、オンプレミスのデータセンターやクラウドで利用できる大規模分散データ処理のために設計された統合エンジンです。

![Screenshot 2023-08-16 at 13.04.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/01c2bbad-d03d-d284-5cee-ba214117eb69.png)

Sparkは中間結果をメモリーで保持するのでHadoop MapReduceよりもはるかに高速になります。機械学習(MLlib)、インタラクティブなクエリーのためのSQL(Spark SQL)、リアルタイムデータを操作するためのストリーム処理(構造化ストリーミング)、グラフ処理(GraphX)で利用可能なAPIを持つライブラリと連携します。

Sparkの設計哲学は4つのキーとなる特性を中心としています:

- **スピード**: Sparkではさまざまな方法でスピードを追求しています。大量の安価なハードウェアを活用することで、効率的なマルチスレッド、並列処理を実現しています。すべての中間結果がメモリーに保持され、ディスクのI/Oを削減することで高いパフォーマンスを実現しています。
- **使いやすさ**: データフレームやデータセットのような高レベルのデータ構造がベースとしているRDD(Resilient Distributed Dataset)を用いて、シンプルな論理的データ構造を提供することでシンプルさを実現しています。一連の*トランスフォーメーション*と*アクション*を提供することで、慣れ親しんだ言語でビッグデータアプリケーションの構築で活用できるシンプルなプログラミングモデルの恩恵を受けることができます。
- **モジュール性**: Sparkのオペレーションは、さまざまなワークロードに適用することができ、サポートされるプログラミング言語で表現することができます。Scala、Java、SQL、Rがサポートされています。また、SparkはSpark SQL、構造化ストリーミング、機械学習のためのMLlib、グラフ解析のためのGraphXのようなコアコンポーネントのAPIを持つライブラリと統合されています。
- **拡張可能性**: Sparkはストレージよりも自身の高速な並列処理エンジンにフォーカスしています。ストレージと計算資源を持つApache Hadoopと異なり、Sparkではそれらを分離しています。これは、Apache Hadoop、Apache Cassandra、Apache HBase、MongoDB、Apache Hive、RDBMSなどのさまざまなソースに格納されているデータを読み込み、すべてをメモリーで処理するためにSparkを活用できることを意味します。また、SparkのDataFrameReaderやDataFrameWriterは、Apache Kafka、Kinesis、Azure Storage、Amazon S3などのソースからデータを読み込み、それらを操作できるように論理的なデータ抽象化レイヤーに読み込むように拡張することができます。
![](https://www.oreilly.com/api/v2/epubs/9781492050032/files/assets/lesp_0102.png)

# Apache Sparkの基本アーキテクチャ

アーキテクチャを理解するには、ドライバーとエグゼキューター(ワーカー)から構成されるSparkクラスターを理解することが重要です。ドライバーが司令塔、エグゼキューターが実際の処理を担います。これによって、並列処理を実現します。

![Screenshot 2023-08-16 at 11.28.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/28c19a2f-4f8d-40ad-bda5-b80310f6b712.png)

- クラスターマネージャが物理マシンを管理し、Sparkアプリケーションにリソースを割り当てます。
- ドライバープロセスは、タスクを完了するためにエグゼキューターでドライバープログラムを実行することに責任を持ちます。
- エグゼキューターはSparkコードを実行します。

**Sparkクラスター**
![Screenshot 2023-08-16 at 11.30.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/41ddcbac-808c-19d4-9931-68a70c914907.png)

# Sparkにおける処理

処理の流れを動画にしました。以下の用語の理解の助けになるかと。

- シェアリソース (Shared resources)
- 並列化 (Parallelization)
- パーティション (Partitions)
- ジョブ、ステージ、タスク (Jobs、 Stages、 Tasks)
- ドライバー (Drivers)
- エグゼキューター (Executors)
- クラスター、ノード (Clusters、 Nodes)
- スロット、コア、スレッド (Slots、 Cores、 Threads)

<iframe width="560" height="315" src="https://www.youtube.com/embed/PU4AluR0ylY" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" allowfullscreen></iframe>

こちらの記事も参考になります。

https://qiita.com/uryyyyyyy/items/ba2dceb709f8701715f7

Sparkアプリケーション(ユーザーのコード)が実行されると、処理対象のデータを分担できるようにデータを分割することで**パーティション**が作成され、ドライバーはSpark**ジョブ**を生成します。さらにジョブを**ステージ**に分割し、最終的には**タスク**に分割します。このタスクとパーティションをエグゼキューターに割り当て、エグゼキューターは割り当てられたパーティションに対してタスクを実行(execute)します。これがエグゼキューターの名前の由来です。なお、パーティションのサイズを明示的に指定することも可能です。パフォーマンスの最適化を行う際には、このパーティションサイズが鍵となることが多いです。

![Screenshot 2023-08-16 at 11.31.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0e212d1e-6df7-bbed-90e3-a382c0bae5c2.png)

:::note
**注意**
ここで述べている**タスク**は集計やフィルタリングなどのデータ処理に限定されません。Pythonの関数で記述できるものであれば基本何でも大丈夫です。このため、機械学習モデルのトレーニングや推論、自然言語処理なども分散処理することが可能です。
:::

Sparkのパーティションに関してはこちらも参照ください。

https://qiita.com/taka_yayoi/items/bb32d5b7abafd600af98

https://qiita.com/taka_yayoi/items/8d964a4949f41e2bf6bd

## Sparkセッション

![Screenshot 2023-08-16 at 11.43.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f4c77811-d35e-96b7-d7cb-90f49bcea6da.png)

- SparkSessionはすべてのデータフレームAPIの機能に対する単一のエントリーポイントです。
- Databricksノートブックでは自動的に変数`spark`が作成されます。`spark...`という形で各種メソッドを呼び出して処理を行います。

以下のようなメソッドがあります。

| メソッド | 説明 |
|:--|:--|
|sql   | 指定されたクエリーの結果を表現するデータフレームを返却。  |
| table  | 指定されたテーブルをデータフレームとして返却。  |
| read  |  データフレームとしてデータを読み込む際に使用できるDataFrameReaderを返却。 |
| range  | startからend(含まない)の範囲とステップ値、パーティション数を持つ要素を含むカラムを持つデータフレームを生成。  |
| createDataFrame  | タプルのリストからデータフレームを作成、主にテストで使用。  |

# Sparkデータフレーム

Sparkでのデータ操作においても、pandasやRと同様にデータフレームを用いることが一般的です。Sparkデータフレームは、名前付きカラムでグルーピングされる分散データコレクションです。

![Screenshot 2023-08-16 at 11.45.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bd1af9ce-e922-d2d0-4fb9-908ea0d387cf.png)

- データはストレージの(データ)パーティションに分散されています。このパーティションはSparkパーティションとは別物です。日付のフォルダなどに格納されている状態のことを指しています。
- Sparkはデータを**パーティション**と呼ばれる塊に分割します。
- Sparkデータフレームによって、物理パーティションにあるデータに対して高レベルの変換処理を適用することができます。

![Screenshot 2023-08-16 at 11.47.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/42ddda11-d150-32e1-7d14-bb1d7a02fcf8.png)

## データフレームに対するトランスフォーメーションとアクション

Sparkにおけるデータフレームの操作においては、**トランスフォーメーション**と**アクション**を理解することも重要です。Sparkでは大規模データの高速処理を実現するために実行計画を立て、最適化を行うため **遅延評価(lazy evaluation)** のアプローチをとっています。逐次実行ではなく、処理のチェーンを組み立てた後に最適化を行うことで、最適なパフォーマンスを実現しています。処理のチェーンを組み立てる際に使用するメソッドが**トランスフォーメーション**であり、トランスフォーメーションを実行するだけでは実際には処理は行われません。そして、組み立てられた処理のチェーンを最適化して実行を行うメソッドが**アクション**となります。

以下の図の右上がトランスフォーメーションのメソッドであり、右下がアクションです。データを取得する、保存するという場合に呼び出すメソッドがアクションとなっています。
![Screenshot 2023-08-16 at 11.57.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9176fcdc-395e-ed3b-4240-968c6c39abd6.png)

# サンプル

サンプルノートブックはこちらです。

https://github.com/taka-yayoi/public_repo_2/blob/master/spark_tutorial/Databricks%E3%81%A8Apache%20Spark%E3%81%AE%E3%81%94%E7%B4%B9%E4%BB%8B.py

Databricks上でSparkを実行していきます。

Sparkセッションを確認します。

```py
spark
```

情報が表示されます。
![Screenshot 2023-08-16 at 12.08.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e92c44fb-239b-56a1-a92b-19b5ae5ad8b8.png)

メソッド`range`を使ってデータフレームを作成し処理を行っていきます。

```py
firstDataFrame = spark.range(1000000)
print(firstDataFrame)
```

`range`はトランスフォーメーションなのでこの時点では処理実行されません。データフレームのスキーマのみが表示されます。

```
DataFrame[id: bigint]
```

さらにトランスフォーメーションを宣言していきます。

```py
# トランスフォーメーションの例
# IDカラムを選択し、2倍します
secondDataFrame = firstDataFrame.selectExpr("(id * 2) as value")
```

アクション`take`を呼び出して処理を実行します。

```py
# アクションの例
# firstDataFrameの最初の５行を取得します
print(firstDataFrame.take(5))
# secondDataFrameの最初の５行を取得します
print(secondDataFrame.take(5))
```

結果が返ってきます。

```
[Row(id=0), Row(id=1), Row(id=2), Row(id=3), Row(id=4)]
[Row(value=0), Row(value=2), Row(value=4), Row(value=6), Row(value=8)]
```

なお、この際には**Sparkジョブ**と表示され、Sparkのジョブが実行されたことを確認することができます。
![Screenshot 2023-08-16 at 12.11.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/84c92bf3-5889-dc3c-59bc-78226ad55876.png)

Databricksの`display`関数もアクションとして動作します。

```py
# display()コマンドでsecondDataFrameを表示します
display(secondDataFrame)
```
![Screenshot 2023-08-16 at 12.12.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/de4b35e5-cebb-25d6-8288-c660281885e3.png)

もう少し複雑な処理を行ってみます。ストレージに格納されているcsvファイルを読み込みます。`read`メソッドとフォーマット、オプションを指定してデータフレームに読み込みます。

```py
dataPath = "/databricks-datasets/Rdatasets/data-001/csv/ggplot2/diamonds.csv"
diamonds = spark.read.format("csv")\
  .option("header","true")\
  .option("inferSchema", "true")\
  .load(dataPath)
  
# inferSchemaはデータを読み込んで自動的にカラムの型を識別するオプションです。データを読み込む分のコストがかかります。
```

データを確認します。

```py
display(diamonds)
```

ダイアモンドのデータです。
![Screenshot 2023-08-16 at 12.14.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/312f57a0-36ec-d526-e41c-d2442d58f28c.png)

このデータに対して集計、joinを行います。`groupBy`や`join`を使います。個人的にはpandasやRのdplyrと大差ないと思います。これらもトランスフォーメーションなので処理は実行されません。

```py
df1 = diamonds.groupBy("cut", "color").avg("price") # シンプルなグルーピング

df2 = df1\
  .join(diamonds, on='color', how='inner')\
  .select("`avg(price)`", "carat")
# シンプルなjoin及び列の選択
```
![Screenshot 2023-08-16 at 12.16.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/dd3789ab-5ee1-ecdb-7ede-698ead70143b.png)

アクションの`count`を実行して処理を行います。

```py
df2.count()
```

```
Out[11]: 269700
```

## Pandas API on Spark

ここまではPythonのSpark APIであるpysparkを使ってきていますが、敷居が高いというのは否定できません。その場合には、pandas APIでSparkデータフレームを操作できるPandas API on Spark(旧Koalas)から入るのが良いかもしれません。

```py
import numpy as np
import pandas as pd
import pyspark.pandas as ps
```

```py
# Pandas API on Sparkのデータフレームの作成
kdf = ps.DataFrame({'A': np.random.rand(5),
                    'B': np.random.rand(5)})
```

pandasのAPIを用いながらSparkの並列処理を活用することができます。

```py
kdf.sort_index()
```
![Screenshot 2023-08-16 at 12.20.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9737a665-4929-8732-0bbd-1b6d43d26ef3.png)

```py
kdf.head(2)
```
![Screenshot 2023-08-16 at 12.21.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cafd2ec1-b69c-174f-7b54-126703e2d634.png)

ただし、注意点もありますのでこちらの記事も一読ください。

https://kakehashi-dev.hatenablog.com/entry/2022/12/24/090000

個人的には、大規模データを取り扱う際には基本はPySparkで処理を行い、pandasのメソッドが必要な際には、データを絞った上で明示的にセルを分けてpandasデータフレームに変換して処理を行い、Sparkデータフレームに戻すという方法を取っています。どのタイプのデータフレーム(pandas/Pandas API on Spark/Spark)を取り扱っているのかが混乱するので。

# なぜ、DatabricksのSparkを使うのか？

Apache Sparkは完全にオープンソースなソフトウェアですが、なぜDatabricks上で使った方が良いのでしょうか？いくつかの理由があります。宣伝めいてすみません。

![Screenshot 2023-08-16 at 12.04.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f996d1ae-49a0-599e-0d0f-c6bf83979c35.png)

1. Databricksに搭載されているSparkはクラウドや[Delta Lake](https://www.databricks.com/jp/product/delta-lake-on-databricks)に最適化されており、非常に高速かつ高いコスト効率を実現します。
1. 機械学習や大規模データエンジニアリングを行う際にシームレスにSparkを活用することができ、生産性の向上が見込めます。
1. Apache Sparkのオリジナルクリエーターが創業した企業がDatabricksであり、数多くのコミッターを抱えていることに加え、エキスパートのサポートを得ることができます。

大規模言語モデルの勃興など、大規模データを取り扱うケースが増えてきている中、すべてのデータをメモリに載せないと処理できないpandasではすぐに限界に直面することになると思います。Sparkに移行することで、水平スケールによる並列処理や[インテリジェントな最適化](https://qiita.com/taka_yayoi/items/38e59a224dfa40994638)の恩恵を受けられるようになり、分析や処理の幅も広がるかと思います。是非、ご活用を検討ください。イベントや勉強会なども開催していく予定です！

Sparkに興味がある方はこちらの記事もどうぞ。

- [PySparkことはじめ](https://qiita.com/taka_yayoi/items/a7ee6287031374efa88a)
- [Sparkを用いたサンフランシスコ消防署の通報データの分析](https://qiita.com/taka_yayoi/items/4ec54d7ba37c71552322)
- [3つのApache Spark APIの物語：RDD対データフレーム、データセット](https://qiita.com/taka_yayoi/items/cc0cebd566515b7ee70c)
- [SparkにおけるPandas API](https://qiita.com/taka_yayoi/items/db8e1ea52afe5c282c94)
- [サンプルを通じたPandasとPySparkデータフレームの比較](https://qiita.com/taka_yayoi/items/5f1ab08c6a97ba96bb1f)
- [可視化を通じたApache Sparkアプリケーションの理解](https://qiita.com/taka_yayoi/items/7d63795bbe6a0b443468)
- [DatabricksでApache Sparkコードを高速にデバッグする7つのTips](https://qiita.com/taka_yayoi/items/6fa8de955e902ee9a365)

### Databricksクイックスタートガイド

[Databricksクイックスタートガイド](https://www.amazon.co.jp/dp/B09V1YXFVQ/)


### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)