E-MapReduce:Lire et écrire des données dans MySQL

Créez une session SQL. Dans la section Sessions, créez une session SQL et sélectionnez une connexion réseau préconfigurée de type Normal Network Connection. Pour plus d'informations, consultez la rubrique Créer une session SQL.

Créez une tâche SQL. Dans la section Development, créez une tâche Spark SQL et exécutez le code SQL suivant.

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

Le tableau suivant décrit les paramètres de l'instruction SQL.

Si la requête renvoie le contenu de la table, la connexion est établie avec succès.

Insérez des données. Exécutez la commande suivante pour insérer des données dans la table MySQL.

INSERT INTO test VALUES(4, 'd'),(5, 'e');
SELECT * FROM test;

Si la requête renvoie les données insérées, l'opération d'écriture a réussi.

Créez une session Notebook. Dans la section Sessions, créez une session Notebook et sélectionnez une connexion réseau préconfigurée de type Normal Network Connection. Pour plus d'informations, consultez la rubrique Créer une session Notebook.

Créez une tâche Notebook. Dans la section Development, créez une tâche de type Interactive Development > Notebook et exécutez le code Python suivant pour effectuer un test.

df = spark.read \
  .format("jdbc") \
  .option("url", "jdbc:mysql://<jdbc_url>") \
  .option("dbtable", "<db>.<table>") \
  .option("user", "<username>") \
  .option("password", "<password>") \
  .load()
df.show()

Si la sortie affiche le contenu de la table, 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 MySQL.

df = spark.createDataFrame([(6, 'f'), (7, 'g')], ["id", "name"])
df.write \
  .format("jdbc") \
  .mode("append") \
  .option("url", "jdbc:mysql://<jdbc_url>") \
  .option("dbtable", "<db>.<table>") \
  .option("user", "<username>") \
  .option("password", "<password>") \
  .save()
df.show()

Le paramètre mode("append") spécifie le mode d'écriture par ajout, qui ajoute des données à la table sans écraser les données existantes.

Si la sortie affiche les données insérées, l'opération d'écriture a réussi.

Compilez et empaquetez le code de test Scala suivant 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:mysql://<jdbc_url>")
      .option("dbtable", "<db>.<table>")
      .option("user", "<username>")
      .option("password", "<password>")
      .save()
    spark.read.format("jdbc")
      .option("url", "jdbc:mysql://<jdbc_url>")
      .option("dbtable", "<db>.<table>")
      .option("user", "<username>")
      .option("password", "<password>")
      .load()
      .show()
    spark.stop()
  }
}

Créez une tâche par lots. Dans la section Development, créez une tâche de type Application > JAR et configurez les paramètres suivants. Pour plus d'informations, consultez la rubrique Développer une tâche par lots ou en streaming.

• 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.

Vérifiez le résultat. Une fois la tâche exécutée, cliquez sur Log Exploration dans la section Execution Records. Dans l'onglet Stdout du journal Driver Log, vous pouvez consulter le contenu de la table MySQL.