Read and write PostgreSQL - E-MapReduce - Centre de documentation Alibaba Cloud

Serverless Spark se connecte à PostgreSQL via un connecteur JDBC intégré. Lisez et écrivez des données à l'aide d'une session SQL, d'un job par lot Spark ou d'une session Notebook.

Prérequis

Vous avez créé un espace de travail Serverless Spark. Pour plus d'informations, consultez la rubrique Créer un espace de travail.
Vous avez créé une instance PostgreSQL.

Vous pouvez utiliser une instance PostgreSQL auto-gérée ou un service de base de données Alibaba Cloud tel que RDS for PostgreSQL ou PolarDB for PostgreSQL.

Cette rubrique utilise RDS for PostgreSQL comme exemple. Pour plus d'informations, consultez la rubrique Créer rapidement une instance RDS for PostgreSQL.

Notes d'utilisation

Exigences relatives au pilote JDBC :
- Les versions suivantes du moteur Serverless Spark incluent le pilote JDBC PostgreSQL (version 42.7.6) par défaut ; vous n'avez donc pas besoin de le fournir.
  - esr-4.4.0 et versions ultérieures
  - esr-3.4.0 et versions ultérieures
  - esr-2.8.0 et versions ultérieures
- Si vous utilisez une version du moteur antérieure aux versions listées ci-dessus, téléchargez le pilote JDBC PostgreSQL et chargez-le sur OSS. Ensuite, dans Sessions > Spark Configuration, ajoutez le paramètre suivant.
```
spark.emr.serverless.user.defined.jars oss://<bucket>/path/to/postgresql-<version>.jar
```
Configuration réseau : assurez-vous que Serverless Spark peut se connecter à votre service PostgreSQL. Pour les détails de configuration, consultez la rubrique Établir la connectivité réseau entre EMR Serverless Spark et d'autres VPC.

Remarque
Lors de la configuration des règles du groupe de sécurité, ouvrez uniquement les ports requis. Cet exemple nécessite le port TCP 5432.

Procédure

Méthode 1 : Utiliser une session SQL

Créez une session SQL. Dans Sessions, créez-en une et sélectionnez une Normal Network Connection préconfigurée. Pour plus d'informations, consultez la rubrique Créer une session SQL.

Créez une tâche SQL. Dans Development, créez une tâche de type SparkSQL et exécutez le code SQL suivant pour tester la connexion.

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;

Le tableau suivant décrit les paramètres.

Paramètre	Description
`url`	La chaîne de connexion JDBC, qui inclut l'hôte PostgreSQL, le port et le nom de la base de données. Utilisez le format `jdbc:postgresql://<jdbc_url>/<database>` et remplacez les espaces réservés par vos valeurs réelles.
`dbtable`	La table à lire, au format `<schema>.<table>`.
`user`	Le nom d'utilisateur de la base de données PostgreSQL. Remarque L'utilisateur doit disposer des autorisations de lecture sur la table cible.
`password`	Le mot de passe de la base de données PostgreSQL.

Si le contenu de la table s'affiche correctement, la connexion est établie avec succès.

Insérez des données. Exécutez le code SQL suivant pour insérer des données dans la table.
```
INSERT INTO test VALUES(4, 'd'),(5, 'e');
SELECT * FROM test;
```
Si les données insérées sont renvoyées correctement, l'opération d'écriture a réussi.

Méthode 2 : Utiliser une session Notebook

Créez une session Notebook. Dans Sessions, créez-en une et sélectionnez une Normal Network Connection préconfigurée. Pour plus d'informations, consultez la rubrique Créer une session Notebook.
Créez une tâche Notebook. Dans Development, créez une tâche de type Interactive Development > Notebook et exécutez le code Python suivant pour tester la connexion.
```
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()
```
Si le contenu de la table s'affiche correctement, la connexion est établie avec succès.

Insérez des données. Exécutez le code suivant pour insérer des données dans la 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()

Le paramètre mode("append") ajoute des données à la table sans écraser les lignes existantes.

Si les données insérées sont renvoyées correctement, l'opération d'écriture a réussi.

Méthode 3 : Utiliser un job par lot Spark

Écrivez le code de test. Compilez le code Scala suivant et empaquetez-le dans un fichier 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()
  }
}

Créez un job par lot. Dans Development, créez une tâche Application > JAR et configurez les paramètres suivants.
- Main JAR Resource : sélectionnez ou saisissez le chemin d'accès au fichier JAR empaqueté.
- Main Class : spark.test.Main.
- Normal Network Connection : sélectionnez une connexion réseau préconfigurée.
Consultez les résultats. Une fois le job terminé, cliquez sur Log Exploration dans la section Execution Records. Dans l'onglet Stdout du Driver Log, vous pouvez consulter le contenu de la table PostgreSQL.