E-MapReduce:MySQL からのデータの読み取りと MySQL へのデータの書き込み

SQL セッションを作成します。[セッション管理] でセッションを作成し、事前構成済みの [ネットワーク接続] を選択できます。詳細については、「SQL セッションの作成」をご参照ください。

[データ開発] で [SparkSQL] タスクを作成し、次の 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;

次の表にパラメーターを示します。

テーブルの内容が正しく返された場合、接続は成功です。

データを挿入します。 次のコマンドを使用して、MySQL テーブルにデータを挿入します。

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

挿入したデータをクエリできれば、書き込み操作は成功です。

Notebook セッションを作成します。[セッション管理] で Notebook セッションを作成し、事前構成済みの [ネットワーク接続] を選択できます。詳細については、「Notebook セッションの作成」をご参照ください。

[データ開発] で [インタラクティブ開発] > [Notebook] タスクを作成し、次の Python コードを使用して接続をテストします。

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

テーブルの内容が正しく返された場合、接続は成功です。

データを挿入します。 次のコードを使用して、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()

mode("append") パラメーターは、既存のデータを上書きまたは削除するのではなく、新しいデータをターゲットテーブルに追加することを指定します。

挿入したデータが正常に取得された場合、書き込み操作は成功です。

テストコードを記述します。 次の Scala コードをコンパイルし、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()
  }
}

バッチジョブを作成します。 [データ開発] で [バッチジョブ] > [JAR] タスクを作成します。次に、次のパラメーターを構成します。構成の詳細については、「バッチジョブまたはストリーミングジョブの開発」をご参照ください。

• [メイン JAR リソース]: JAR ファイルへのパスを選択または入力します。

• [メインクラス] は spark.test.Main です。

• [ネットワーク接続]: 既存のネットワーク接続を選択します。

タスク完了後に検証結果を表示するには、[実行記録] セクションの [ログ調査] をクリックします。[ドライバーログ] の [Stdout] タブで、MySQL データベース内の対応するテーブルの内容を表示できます。