OCI Data Flowを使ってオブジェクト・ストレージからADBへデータをロードする -- Part 1

本ブログは、[オラクル・クラウドの個人シリーズ・ブログ](https://qiita.com/letian/items/10152a4fec813c1b7d98)の1つです。

# 初めに
Oracle Cloud Infrastructure（OCI）Data Flowは、[Apache Spark](https://spark.apache.org/) ™アプリケーションを実行するためのフルマネージド・サービスです。特に大規模なデータの処理に向きます。

OCIオブジェクト・ストレージ上のファイルを読み込み、Autnomous DBへのデータロード処理を例にして、OCI Data Flowでの実現方法を三回を分けて紹介したいと思います。

- <strong>Part 1 (本文)</strong>: Apache Sparkを使って、ローカル環境でPythonプログラムを実行させる方法。
- [Part 2](https://qiita.com/liu-wei/items/eb3c57ed6da656e581cd): アプリケーションをOCIへデプロイし、実行させる方法。
- [Part 3](https://qiita.com/liu-wei/items/6a20d4b750f8610ae895): プライベート・エンドポイントを経由し、ADBへデータをロードする方法。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1121622/4f549f24-0ade-482e-232d-04f7498dec43.png", width="450">

__検証環境__
VM: Oracle Linux 8, VM.Standard.E4.Flex (1 OCPU, 16GB)
ADB タイプ: Autonomous Data Warehouse (パブリック・アクセス)
Java バージョン: 11
Python バージョン: 3.8
Apache Spark バージョン: 3.3.2

__ステップ__
- [1. OCI側の準備](#1-oci側の準備)
  - [バケットとCSVファイル](#バケットとcsvファイル)
  - [ADBインスタンスと専用ユーザー](#adbインスタンスと専用ユーザー)
  - [ADB Walletファイル](#adb-walletファイル)
- [2. ローカル環境のセットアップ](#2-ローカル環境のセットアップ)
  - [Java 11 のインストール](#java-11-のインストール)
  - [Python のインストール](#python-のインストール)
  - [Spark のインストール](#spark-のインストール)
  - [JDBCドライバーのダウンロード](#jdbcドライバーのダウンロード)
  - [OCI HDFS Connector のダウンロード](#oci-hdfs-connector-のダウンロード)
  - [Python SDK のインストール](#python-sdk-のインストール)
  - [OCI構成ファイルの準備](#oci構成ファイルの準備)
- [3. Pythonプログラムの作成](#3-pythonプログラムの作成)
- [4. ローカル環境での実行](#4-ローカル環境での実行)

# 1. OCI側の準備

## バケットとCSVファイル
バケット（一個）を用意して、読込対象のCSVファイルをその中に保存します。この例では、次のようなファイルを利用します（ヘッダー付き、郵便番号と住所を記載）。
```
zipcode,address1,address2,address3
0600000,北海道,札幌市　中央区,以下に掲載がない場合
0640941,北海道,札幌市　中央区,旭ケ丘
0600041,北海道,札幌市　中央区,大通東
0600042,北海道,札幌市　中央区,大通西（１～１９丁目）
0640820,北海道,札幌市　中央区,大通西（２０～２８丁目）
...中略...
```

## ADBインスタンスと専用ユーザー
ADBインスタンス（一個）を新規作成します。この例では、Autonomous Data Warehouse (ADW)を利用します。

ロード処理の専用ユーザー(例：`DFUSER01`)を作成しておきます（ロール`DWROLE`を付与）。

パスワードをOCI Vaultのシークレットに保存します。保存後、<strong>シークレットのOCID</strong>をメモしてください（例: `ocid1.vaultsecret.oc1.ap-tokyo-1.<secret_ocid>`）。

## ADB Walletファイル
OCIコンソールから、Walletファイルをダウンロードします。圧縮ファイルをそのままバケットに保存してください（解凍不要）。

# 2. ローカル環境のセットアップ
## Java 11 のインストール
コマンド：`sudo dnf install java-11-openjdk`
バージョンチェック：`java -version`

```bash
[opc@linux8-java11-data-flow ~]$ sudo dnf install java-11-openjdk
...
Complete!
[opc@linux8-java11-data-flow ~]$ java -version
openjdk version "11.0.18" 2023-01-17 LTS
OpenJDK Runtime Environment (Red_Hat-11.0.18.0.10-2.el8_7) (build 11.0.18+10-LTS)
OpenJDK 64-Bit Server VM (Red_Hat-11.0.18.0.10-2.el8_7) (build 11.0.18+10-LTS, mixed mode, sharing)
[opc@linux8-java11-data-flow ~]$
```
## Python のインストール
インストール：`sudo dnf module install python38`
バージョンチェック：`python3.8 --version`
実行環境をPython3.8に指定：`sudo alternatives --set python3 /usr/bin/python3.8`

## Spark のインストール
ダウンロードURL：https://spark.apache.org/downloads.html
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1121622/e9c904b1-992b-9cf8-333f-4b8556cf0b16.png)
ファイル：`spark-3.3.2-bin-hadoop3.tgz`
ローカルの展開先：`/home/opc/spark/` (事前に作成しておく)
コマンド例：`tar xvf spark-3.3.2-bin-hadoop3.tgz -C /home/opc/spark`

__PATH変数の編集__
```
export Spark_Home=/home/opc/spark/spark-3.3.2-bin-hadoop3
export PATH=${Spark_Home}/bin:$PATH
```

__Spark Shell の起動__
`spark-shell`で正常に起動できるのかを確認します。
```
[opc@linux8-java11-data-flow ~]$ spark-shell
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/04/24 11:28:57 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://linux8-java11-data-flow.publicsubnet1.vcn1.oraclevcn.com:4040
Spark context available as 'sc' (master = local[*], app id = local-1682335738728).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.3.2
      /_/

Using Scala version 2.12.15 (OpenJDK 64-Bit Server VM, Java 11.0.18)
Type in expressions to have them evaluated.
Type :help for more information.

scala>
```
Quit コマンド: `CTRL + D`

## JDBCドライバーのダウンロード
ダウンロードURL：https://www.oracle.com/database/technologies/appdev/jdbc-downloads.html
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1121622/b5ff37e1-9f84-b337-d21e-fa853b0c8cfb.png)
ファイル：`ojdbc11-full.tar.gz`
ローカルの展開先：`/home/opc/download/ojdbc/` (事前に作成しておく)

## OCI HDFS Connector のダウンロード
ダウンロードURL：https://github.com/oracle/oci-hdfs-connector/releases
ファイル名：`oci-hdfs.zip`
ローカルの展開先：`/home/opc/download/oci-hdfs`

展開後、下記ファイルをsparkのjarsの下に保存してださい。
保存先：`/home/opc/spark/spark-3.3.2-bin-hadoop3/jars`
- oci-hdfs/lib/oci-hdfs-full-3.3.4.0.1.1.jar
- oci-hdfs/third-party/lib/bcpkix-jdk15on-1.70.jar
- oci-hdfs/third-party/lib/bcprov-jdk15on-1.70.jar

## Python SDK のインストール
コマンド：`sudo pip3 install -U oci`

## OCI構成ファイルの準備
オブジェクト・ストレージにアクセスするため、認証情報を記載する<strong>OCI構成ファイル</strong>と<strong>APIキーファイル(PEM)</strong>は必要です。

ファイル作成方法の詳細は、次の記事をご参照ください。
[「OCIカスタム・メトリックでディスク使用率を監視」](https://qiita.com/liu-wei/items/3f5593eab06c5885a83a)の「STEP1. 事前準備」

ファイルの保存先：`/home/opc/.oci/config`
（構成ファイルとPEMキーファイルの保存先を別々にしてもOKです。）

# 3. Pythonプログラムの作成
[Github](https://github.com/oracle-samples/oracle-dataflow-samples)に、Python、JavaとScalaの３種類のサンプルがあります。今回は、Pythonの[サンプルコード](https://github.com/oracle-samples/oracle-dataflow-samples/tree/main/python/loadadw)を例にして検証します。

ソースの格納先：`/home/opc/python/loadadw.py`
編集箇所は、以下となります。
```python
    # TODO: Set all these variables.
    INPUT_PATH = "oci://<bucket>@<object_storage_namespace>/<filename>.csv"
    PASSWORD_SECRET_OCID = "ocid1.vaultsecret.oc1.ap-tokyo-1.<secret_ocid>"
    TARGET_TABLE = "ZIPCODEJP"
    TNSNAME = "adwdf_high"
    USER = "DFUSER01"
    WALLET_PATH = "oci://<bucket>@<object_storage_namespace>/<wallet_file>.zip"
```

オブジェクト・ストレージ・ネームスペースは、OCIコンソールから確認できますが、OCI CLIの`oci os ns get`でも簡単に取得できます。（テナント名ではないので、ご注意を）

__処理流れ__
CSVファイルの読込 → ADB Walletファイルの取得と解凍 → シークレットからパスワードの取得 → ADBへの書込み

# 4. ローカル環境での実行
ターゲット・テーブルを予め作成しなくてもよいです。実行後自動に作成されます。列名は、CSVファイルのヘッダーと同様です。

実行コマンド：
`spark-submit --jars /home/opc/download/ojdbc/ojdbc11.jar /home/opc/python/loadadw.py`

出力ログの抜粋：
```bash
[opc@linux8-java11-data-flow ]$ spark-submit --jars /home/opc/download/ojdbc/ojdbc11.jar /home/opc/python/loadadw.py
23/04/24 12:46:15 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
23/04/24 12:46:17 INFO SparkContext: Running Spark version 3.3.2
23/04/24 12:46:17 INFO ResourceUtils: ==============================================================
23/04/24 12:46:17 INFO ResourceUtils: No custom resources configured for spark.driver.
23/04/24 12:46:17 INFO ResourceUtils: ==============================================================
23/04/24 12:46:17 INFO SparkContext: Submitted application: DataFlow
...中略...
Getting wallet password
Done getting wallet password
Saving processed data to jdbc:oracle:thin:@adwdf_high?TNS_ADMIN=/tmp
23/04/24 12:46:29 INFO BlockManagerInfo: Removed broadcast_0_piece0 on linux8-java11-data-flow.publicsubnet1.vcn1.oraclevcn.com:46195 in memory (size: 34.7 KiB, free: 434.4 MiB)
23/04/24 12:46:29 INFO BlockManagerInfo: Removed broadcast_2_piece0 on linux8-java11-data-flow.publicsubnet1.vcn1.oraclevcn.com:46195 in memory (size: 34.7 KiB, free: 434.4 MiB)
23/04/24 12:46:29 INFO BlockManagerInfo: Removed broadcast_1_piece0 on linux8-java11-data-flow.publicsubnet1.vcn1.oraclevcn.com:46195 in memory (size: 5.9 KiB, free: 434.4 MiB)
23/04/24 12:46:29 INFO FileSourceStrategy: Pushed Filters:
23/04/24 12:46:29 INFO FileSourceStrategy: Post-Scan Filters:
23/04/24 12:46:29 INFO FileSourceStrategy: Output Data Schema: struct<zipcode: string, address1: string, address2: string, address3: string ... 2 more fields>
...中略...
23/04/24 12:46:32 INFO DAGScheduler: ResultStage 1 (jdbc at NativeMethodAccessorImpl.java:0) finished in 3.120 s
23/04/24 12:46:32 INFO DAGScheduler: Job 1 is finished. Cancelling potential speculative or zombie tasks for this job
23/04/24 12:46:32 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks have all completed, from pool
23/04/24 12:46:32 INFO TaskSchedulerImpl: Killing all running tasks in stage 1: Stage finished
23/04/24 12:46:32 INFO DAGScheduler: Job 1 finished: jdbc at NativeMethodAccessorImpl.java:0, took 3.130925 s
Done saving processed data to database
...中略...
23/04/24 12:46:33 INFO SparkContext: Successfully stopped SparkContext23/04/24 12:46:33 INFO ShutdownHookManager: Shutdown hook called
23/04/24 12:46:33 INFO ShutdownHookManager: Deleting directory /tmp/spark-eb0988f9-4df7-4154-aa7d-d7fdce1ec33e/pyspark-8995996a-7003-41f4-92ce-d087fa9351ec
23/04/24 12:46:33 INFO ShutdownHookManager: Deleting directory /tmp/spark-89bfb53f-4572-4d7d-be52-eee52e797b9c
23/04/24 12:46:33 INFO ShutdownHookManager: Deleting directory /tmp/spark-eb0988f9-4df7-4154-aa7d-d7fdce1ec33e
[opc@linux8-java11-data-flow ]$
```
全体の処理時間は、およそ18秒で、データ（<font color=blue>12万件以上</font>）の書込み時間は、<font color=blue>4秒</font>未満です。

__テーブルの中身を確認__
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1121622/f3eb3054-af4d-67c7-83ab-e44fe53a6d50.png)

以上です。

----
__関連記事__
[オラクル・クラウドの個人シリーズ・ブログ](https://qiita.com/liu-wei/items/10152a4fec813c1b7d98)
[OCI Data Flowを使ってオブジェクト・ストレージからADBへデータをロードする -- Part 2](https://qiita.com/liu-wei/items/eb3c57ed6da656e581cd)
[OCI Data Flowを使ってオブジェクト・ストレージからADBへデータをロードする -- Part 3](https://qiita.com/liu-wei/items/6a20d4b750f8610ae895)

----
__OCI データ・フロー・ドキュメント__
[製品サイト](https://www.oracle.com/jp/big-data/data-flow/)
[オフィシャル・ドキュメント](https://docs.public.oneportal.content.oci.oraclecloud.com/ja-jp/iaas/data-flow/using/home.htm)
[データ・フロー・チュートリアル](https://docs.oracle.com/ja-jp/iaas/data-flow/data-flow-tutorial/dfs_tut_front.htm)