Baca dan tulis PostgreSQL - E-MapReduce - Alibaba Cloud Documentation Center

Spark menyediakan dukungan native untuk mengakses PostgreSQL melalui konektor JDBC. Anda dapat terhubung ke PostgreSQL untuk membaca dan menulis data menggunakan session SQL, job batch Spark, atau session Notebook.

Prasyarat

Anda telah membuat ruang kerja Serverless Spark. Untuk informasi selengkapnya, lihat Create a workspace.
Anda telah membuat instans PostgreSQL.

Anda dapat menggunakan instans PostgreSQL yang dikelola sendiri atau layanan database Alibaba Cloud seperti RDS for PostgreSQL atau PolarDB for PostgreSQL.

Topik ini menggunakan RDS for PostgreSQL sebagai contoh. Untuk informasi selengkapnya, lihat Quickly create an RDS for PostgreSQL instance.

Catatan penggunaan

Persyaratan driver JDBC:
- Versi engine Serverless Spark berikut mencakup driver JDBC PostgreSQL (versi 42.7.6) secara default, sehingga Anda tidak perlu menyediakannya sendiri.
  - esr-4.4.0 dan yang lebih baru
  - esr-3.4.0 dan yang lebih baru
  - esr-2.8.0 dan yang lebih baru
- Jika Anda menggunakan versi engine yang lebih lama dari versi yang tercantum di atas, unduh PostgreSQL JDBC Driver dan unggah ke OSS. Kemudian, di Sessions > Spark Configuration, tambahkan parameter berikut.
```
spark.emr.serverless.user.defined.jars oss://<bucket>/path/to/postgresql-<version>.jar
```
Konfigurasi jaringan: Pastikan Serverless Spark dapat terhubung ke layanan PostgreSQL Anda. Untuk detail konfigurasi, lihat Establish network connectivity between EMR Serverless Spark and other VPCs.

Catatan
Saat mengonfigurasi aturan security group, buka hanya port yang diperlukan. Contoh ini memerlukan TCP port 5432.

Prosedur

Metode 1: Gunakan session SQL

Buat session SQL. Di Sessions, buat satu session dan pilih Normal Network Connection yang telah dikonfigurasi sebelumnya. Untuk informasi selengkapnya, lihat Create an SQL session.

Buat task SQL. Di Development, buat task bertipe SparkSQL dan jalankan SQL berikut untuk menguji koneksi.

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;

Tabel berikut menjelaskan parameter tersebut.

Parameter	Deskripsi
`url`	String koneksi JDBC, yang mencakup host, port, dan nama database PostgreSQL. Gunakan format `jdbc:postgresql://<jdbc_url>/<database>` dan ganti placeholder dengan nilai aktual Anda.
`dbtable`	Tabel yang akan dibaca, dalam format `<schema>.<table>`.
`user`	Username untuk database PostgreSQL. Catatan User tersebut harus memiliki izin baca pada tabel target.
`password`	Password untuk database PostgreSQL.

Jika konten tabel ditampilkan dengan benar, koneksi berhasil.

Masukkan data. Jalankan SQL berikut untuk memasukkan data ke dalam tabel.
```
INSERT INTO test VALUES(4, 'd'),(5, 'e');
SELECT * FROM test;
```
Jika data yang dimasukkan dikembalikan dengan benar, operasi tulis berhasil.

Metode 2: Gunakan session Notebook

Buat session Notebook. Di Sessions, buat satu session dan pilih Normal Network Connection yang telah dikonfigurasi sebelumnya. Untuk informasi selengkapnya, lihat Create a Notebook session.

Buat task Notebook. Di Development, buat task bertipe Interactive Development > Notebook dan jalankan kode Python berikut untuk menguji koneksi.

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()

Jika konten tabel ditampilkan dengan benar, koneksi berhasil.

Masukkan data. Jalankan kode berikut untuk memasukkan data ke dalam tabel.

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()
df.show()

Parameter mode("append") menentukan mode tambah, yang menambahkan data ke tabel tanpa menimpa baris yang sudah ada.

Jika data yang dimasukkan dikembalikan dengan benar, operasi tulis berhasil.

Metode 3: Gunakan job batch Spark

Tulis kode pengujian. Kompilasi kode Scala berikut dan kemas menjadi file 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()
  }
}

Buat job batch. Di Development, buat task Application > JAR dan konfigurasikan parameter berikut.
- Main JAR Resource: Pilih atau masukkan path ke file JAR yang telah dikemas.
- Main Class: spark.test.Main.
- Normal Network Connection: Pilih koneksi jaringan yang telah dikonfigurasi sebelumnya.
Lihat hasilnya. Setelah job selesai, klik Log Exploration di bagian Execution Records. Di tab Stdout pada Driver Log, Anda dapat melihat isi tabel PostgreSQL.