Read and write PostgreSQL - E-MapReduce - Alibaba Cloud Documentation Center

Prerequisites

You have created a Serverless Spark workspace. For more information, see Create a workspace.
You have created a PostgreSQL instance.

You can use a self-managed PostgreSQL instance or an Alibaba Cloud database service such as RDS for PostgreSQL or PolarDB for PostgreSQL.

This topic uses RDS for PostgreSQL as an example. For more information, see Quickly create an RDS for PostgreSQL instance.

Usage notes

JDBC driver requirements:
- The following Serverless Spark engine versions include the PostgreSQL JDBC driver (version 42.7.6) by default, so you do not need to provide one.
  - esr-4.4.0 and later
  - esr-3.4.0 and later
  - esr-2.8.0 and later
- If you use an engine version earlier than the versions listed above, download the PostgreSQL JDBC Driver and upload it to OSS. Then, in Sessions > Spark Configuration, add the following parameter.
```
spark.emr.serverless.user.defined.jars oss://<bucket>/path/to/postgresql-<version>.jar
```
Network configuration: Ensure Serverless Spark can connect to your PostgreSQL service. For configuration details, see Establish network connectivity between EMR Serverless Spark and other VPCs.

Note
When configuring security group rules, open only the required ports. This example requires TCP port 5432.

Procedure

Method 1: Use an SQL session

Create an SQL session. In Sessions, create one and select a preconfigured Normal Network Connection. For more information, see Create an SQL session.

Create an SQL task. In Development, create a task of type SparkSQL and run the following SQL to test the connection.

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;

The following table describes the parameters.

Parameter	Description
`url`	The JDBC connection string, which includes the PostgreSQL host, port, and database name. Use the format `jdbc:postgresql://<jdbc_url>/<database>` and replace the placeholders with your actual values.
`dbtable`	The table to read, in `<schema>.<table>` format.
`user`	The username for the PostgreSQL database. Note The user must have read permissions on the target table.
`password`	The password for the PostgreSQL database.

If the table content is displayed correctly, the connection is successful.

Insert data. Run the following SQL to insert data into the table.
```
INSERT INTO test VALUES(4, 'd'),(5, 'e');
SELECT * FROM test;
```
If the inserted data is returned correctly, the write operation is successful.

Method 2: Use a Notebook session

Create a Notebook session. In Sessions, create one and select a preconfigured Normal Network Connection. For more information, see Create a Notebook session.

Create a Notebook task. In Development, create a task of type Interactive Development > Notebook and run the following Python code to test the connection.

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

If the table content is displayed correctly, the connection is successful.

Insert data. Run the following code to insert data into the table.

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

The mode("append") parameter specifies the append mode, which adds data to the table without overwriting existing rows.

If the inserted data is returned correctly, the write operation is successful.

Method 3: Use a Spark batch job

Write the test code. Compile the following Scala code and package it into a JAR file.

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

Create a batch job. In Development, create a Application > JAR task and configure the following parameters.
- Main JAR Resource: Select or enter the path to the packaged JAR file.
- Main Class: spark.test.Main.
- Normal Network Connection: Select a preconfigured network connection.
View the results. After the job completes, click Log Exploration in the Execution Records section. On the Stdout tab of the Driver Log, you can view the contents of the PostgreSQL table.