Google Colab の Spark にて Google Cloud Storage と連携する方法

## 概要

Google Colab の Spark から Google Cloud Storage (GCS)と連携する方法を共有します。Google Cloud 上で GCS から初めて、 Google Colab 上の Spark から書き込みと読み込みを実施する手順です。

## 環境構築

### GCS のバケットを作成

今回は us-cental1 リージョンで作成しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6dada41a-abf5-42a5-a0cc-76d814d9ae21.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/654496de-20f5-4988-bd57-45271e51055a.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6273673b-f2f7-46c5-a312-2b31ebca2c4c.png)

`gsutil URI`の値を控えます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/196fa139-87c2-4c69-8870-a2f8d8b29a27.png)

### Service Account の作成

コンソールにて｀IAM & Admin` -> `Service accounts`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0659fd8e-0559-4f65-992a-e93f145d73eb.png)

`+ CREATE SERVICE ACCOUNT`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e6507be1-e864-4099-a936-3209ac0dfb5c.png)

手順に則り Service Account を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1647babf-14d4-4a9c-a559-17b8ed7de7c4.png)

### Service Account の認証情報を取得

作成した Service Account を表示して、`KEYS`タブにある `ADD KEY` -> `Create new key`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/72256048-d649-4e1e-a18e-539c11fee9f3.png)

`JSON`をチェックして、`CREATE`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/22a4e07d-5ddd-4328-aaf7-6713d23788bf.png)

json をダウンロードが開始されるため、ローカル環境に保存します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7c2b421b-fd9f-47f0-8d33-e9d026f80f5e.png)

### GCS のバケットに対する権限を Service Account に付与

作成した GCS のバケットを表示後、`PERMISSONS`タブにある`GRANT ACCESS`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f7d51abf-b4b7-4f6f-ba5c-ef1b05f51fa7.png)

`New principals`に Service Principal のメールアドレスを入力し、`ROLE`にストレージに対する適切な権限（例：`Storage Admin`）を設定した上で`Save`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/047144ea-8456-4123-aa5f-7158510beef6.png)

## Google Colab 上での Spark による操作

### Google Drive 上に Service Account の key ファイルを配置

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b36f5462-5e52-4e15-9d45-b15e41ba600d.png)

### Google Colab 上で新規ノートブック作成後に、 PySpark のバージョンを確認

```bash
!pyspark --version
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f41e1281-72c7-4ee8-9619-65d363c0b58f.png)


### Google Drive を Google Colab のノートブックにマウント

```python
from google.colab import drive
drive.mount('/content/drive')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/85baf40e-bdf2-4451-9795-a79ad73339d3.png)

### 左側の `ファイル`タブを選択し、アップロードした Service Account の key ファイルのパスをコピー

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5f6634c6-41bb-47ce-8211-2a0e7c79e130.png)

### パスを変数にセット

```python
gcs_key_path = "/content/drive/MyDrive/Colab Notebooks/bigquery_iceberg/gcs_key.json"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/25babf21-be20-4baf-90ae-8c84c00a90ab.png)


### SparkSession の定義

```python
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
        .appName("gcs_test")
        .config("spark.jars", "https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop3-latest.jar")
        .config("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")
        .config("google.cloud.auth.service.account.enable", "true")
        .config("spark.hadoop.google.cloud.auth.service.account.enable", "true")
        .config("spark.hadoop.google.cloud.auth.service.account.json.keyfile", gcs_key_path)
        .getOrCreate()
)
spark
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4cb553fe-f1ea-4399-970e-7573de4f48f4.png)

### GCS のパスを変数にセット

```python
gcs_base_path = "gs://manabianqiita"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/21641ccf-4464-4a5d-8420-ec586840dd79.png)

### Spark から GCS に書き込みを実施

```python
data = [("Alice", 34), ("Bob", 45), ("Cathy", 29)]
columns = ["Name", "Age"]
df = spark.createDataFrame(data, columns)

df.write.mode("overwrite").parquet(f"{gcs_base_path}/output_data3")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cafac6ae-2146-4b35-9690-b1173c9a5771.png)


GCS 上で parquet ファイルが書き込まれていることを確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3b9b4901-875d-4a8b-8438-534652a8f55c.png)


### Spark から GCS から読み込みを実施

```python
df = spark.read.parquet(f"{gcs_base_path}/output_data3")
df.show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/41fe34df-b89c-4624-9a30-7aeda412495f.png)