PostgreSQL の読み取りと書き込み - E-MapReduce - Alibaba Cloud ドキュメントセンター

前提条件

Serverless Spark ワークスペースを作成済みであること。詳細については、「Create a workspace」をご参照ください。
PostgreSQL インスタンスを作成済みであること。

自己管理型 PostgreSQL インスタンス、または RDS for PostgreSQL や PolarDB for PostgreSQL などの Alibaba Cloud データベースサービスを使用できます。

このトピックでは、例として RDS for PostgreSQL を使用します。詳細については、「Quickly create an RDS for PostgreSQL instance」をご参照ください。

使用上の注意

JDBC ドライバー要件：
- 次の Serverless Spark エンジンバージョンには、PostgreSQL JDBC ドライバー (バージョン 42.7.6) がデフォルトで含まれているため、提供する必要はありません。
  - esr-4.4.0 以降
  - esr-3.4.0 以降
  - esr-2.8.0 以降
- 上記のバージョンより前のエンジンバージョンを使用する場合は、PostgreSQL JDBC Driver をダウンロードして OSS にアップロードした後、Sessions > Spark Configuration で、次のパラメーターを追加します。
```
spark.emr.serverless.user.defined.jars oss://<bucket>/path/to/postgresql-<version>.jar
```
ネットワーク設定：Serverless Spark が PostgreSQL サービスに接続できることを確認してください。設定の詳細については、「Establish network connectivity between EMR Serverless Spark and other VPCs」をご参照ください。

説明
セキュリティグループルールを設定する際は、必要なポートのみを開放してください。この例では TCP ポート 5432 が必要です。

操作手順

方法 1：SQL セッションの使用

SQL セッションを作成します。Sessions で 1 つ作成し、事前に設定された Normal Network Connection を選択します。詳細については、「SQL セッションを作成する」をご参照ください。

SQL タスクを作成します。Development で、SparkSQL タイプのタスクを作成し、次の SQL を実行して接続をテストします。

CREATE TEMPORARY VIEW test
USING org.apache.spark.sql.jdbc
OPTIONS (
  url 'jdbc:postgresql://<jdbc_url>/<database>',
  dbtable '<schema>.<table>',
  user '<username>',
  password '<password>'
);
SELECT * FROM test;

パラメーターについては、次の表で説明します。

パラメーター	説明
`url`	JDBC 接続文字列です。PostgreSQL のホスト、ポート、データベース名が含まれます。 `jdbc:postgresql://<jdbc_url>/<database>` 形式を使用し、プレースホルダーを実際の値に置き換えます。
`dbtable`	読み取るテーブルです。`<schema>.<table>` 形式で指定します。
`user`	PostgreSQL データベースのユーザー名です。説明ユーザーには、ターゲットテーブルに対する読み取り権限が必要です。
`password`	PostgreSQL データベースのパスワードです。

テーブルの内容が正しく表示されれば、接続は成功です。

データを挿入します。次の SQL を実行してテーブルにデータを挿入します。
```
INSERT INTO test VALUES(4, 'd'),(5, 'e');
SELECT * FROM test;
```
挿入したデータが正しく返されれば、書き込みは成功です。

方法 2：Notebook セッションの使用

Notebook セッションを作成します。 Sessions で 1 つ作成し、事前設定された Normal Network Connection を選択します。詳細については、「Notebook セッションの作成」をご参照ください。
Development で、[対話型開発] > [ノートブック] タイプのノートブックタスクを作成し、次の Python コードを実行して接続をテストします。
```
df = spark.read \
  .format("jdbc") \
  .option("url", "jdbc:postgresql://<jdbc_url>/<database>") \
  .option("dbtable", "<schema>.<table>") \
  .option("user", "<username>") \
  .option("password", "<password>") \
  .load()
df.show()
```
テーブルの内容が正しく表示されれば、接続は成功です。
データを挿入します。次のコードを実行してテーブルにデータを挿入します。
```
df = spark.createDataFrame([(6, 'f'), (7, 'g')], ["id", "name"])
df.write \
  .format("jdbc") \
  .mode("append") \
  .option("url", "jdbc:postgresql://<jdbc_url>/<database>") \
  .option("dbtable", "<schema>.<table>") \
  .option("user", "<username>") \
  .option("password", "<password>") \
  .save()
```
mode("append") パラメーターは追加モードを指定します。このモードでは、既存の行を上書きせずにテーブルへデータを追加します。

書き込みを検証するには、前の手順で示した読み取りコードを再度実行します。新しく挿入されたデータが結果に含まれていることを確認してください。

方法 3：Spark バッチジョブの使用

テストコードを作成します。次の Scala コードをコンパイルし、JAR ファイルにパッケージ化します。

package spark.test
import org.apache.spark.sql.SparkSession
object Main {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("test")
      .getOrCreate()
    val newRows = spark.createDataFrame(Seq((6, "f"), (7, "g"))).toDF("id", "name")
    newRows.write.format("jdbc")
      .mode("append")
      .option("url", "jdbc:postgresql://<jdbc_url>/<database>")
      .option("dbtable", "<schema>.<table>")
      .option("user", "<username>")
      .option("password", "<password>")
      .save()
    spark.read.format("jdbc")
      .option("url", "jdbc:postgresql://<jdbc_url>/<database>")
      .option("dbtable", "<schema>.<table>")
      .option("user", "<username>")
      .option("password", "<password>")
      .load()
      .show()
    spark.stop()
  }
}

バッチジョブを作成します。Development で、Application > JAR タスクを作成し、以下のパラメーターを設定します。
- [Main JAR Resource]：パッケージ化した JAR ファイルのパスを選択または入力します。
- [Main Class]：spark.test.Main。
- [Normal Network Connection]：事前設定済みのネットワーク接続を選択します。
結果を表示します。ジョブが完了したら、Execution Records セクションの Log Exploration をクリックします。Driver Log の [Stdout] タブで、PostgreSQL テーブルの内容を表示できます。