E-MapReduce:Baca dari dan tulis ke MySQL

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

Buat task SQL. Di Development, buat task Spark SQL dan jalankan SQL berikut.

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;

Tabel berikut menjelaskan parameter tersebut.

Jika kueri mengembalikan konten tabel, koneksi berhasil.

Masukkan data. Jalankan perintah berikut untuk memasukkan data ke tabel MySQL.

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

Jika kueri mengembalikan data yang dimasukkan, operasi penulisan berhasil.

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

Buat task Notebook. Di Development, buat task jenis Interactive Development > Notebook dan jalankan kode Python berikut untuk pengujian.

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

Jika output menampilkan konten tabel, koneksi berhasil.

Masukkan data. Jalankan kode berikut untuk memasukkan data ke tabel 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()

Parameter mode("append") mengatur mode penulisan menjadi append, yang menambahkan data ke tabel alih-alih menimpa data yang sudah ada.

Jika output menampilkan data yang dimasukkan, operasi penulisan berhasil.

Kompilasi dan paket kode uji Scala berikut ke dalam 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: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()
  }
}

Buat task batch. Di Development, buat task jenis Application > JAR, dan konfigurasikan parameter berikut. Untuk informasi selengkapnya, lihat Develop a batch or streaming task.

• Main JAR Resource: Pilih atau masukkan path ke file JAR yang telah dipaketkan.

• Main Class: spark.test.Main.

• Normal Network Connection: Pilih koneksi jaringan yang telah dikonfigurasi sebelumnya.

Verifikasi hasilnya. Setelah task dijalankan, klik Log Exploration di bagian Execution Records. Di tab Stdout pada Driver Log, Anda dapat melihat konten tabel dari database MySQL Anda.