Read from and write to MySQL - E-MapReduce - Alibaba Cloud Documentation Center

Apache Spark natively supports accessing MySQL through the JDBC Connector. Serverless Spark automatically loads the MySQL JDBC driver (version 8.0.33) at startup. You can connect to MySQL using an SQL session, a batch task, or a Notebook to read and write data.

Prerequisites

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

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

This topic uses RDS for MySQL as an example. For more information, see Create and configure an RDS for MySQL instance.

Usage notes

Ensure network connectivity between Serverless Spark and MySQL. For configuration details, see Establish network connectivity between EMR Serverless Spark and other VPCs.

Note

When you configure security group rules, open only the required ports. This example requires opening TCP port 3306.

Procedure

Method 1: Using an SQL session

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

Create an SQL task. In Development, create a Spark SQL task and run the following SQL.

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;

The following table describes the parameters.

Parameter	Description
`url`	The JDBC connection string. The format is `jdbc:mysql://<jdbc_url>/`. Replace `<jdbc_url>` with the actual value.
`dbtable`	The database table to read. The format is `<db>.<table>`. This topic uses `test_mysql_db.test` as an example.
`user`	The username for the MySQL database. Note The user must have read permission on the target table.
`password`	The password for the MySQL database.

If the query returns table content, the connection is successful.

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

Method 2: Using a Notebook session

Create a Notebook session. In Sessions, create a Notebook session 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 the Interactive Development > Notebook type and run the following Python code to 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()

If the output shows the table content, the connection is successful.

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

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

The mode("append") parameter sets the write mode to append, which adds data to the table instead of overwriting existing data.

If the output shows the inserted data, the write operation is successful.

Method 3: Using a Spark batch task

Compile and package the following Scala test code 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: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()
  }
}

Create a batch task. In Development, create a task of the Application > JAR type, and configure the following parameters. For more information, see Develop a batch or streaming task.
- 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.
Verify the result. After the task is executed, click Log Exploration in the Execution Records section. On the Stdout tab of Driver Log, you can view the table contents from your MySQL database.