E-MapReduce:Doris からの読み込みと Doris への書き込み

SQL セッションを使用した Doris テーブルの読み込み

1. SQL セッションを作成します。詳細については、「SQLセッションの管理」をご参照ください。

   セッションを作成する際、Engine Version ドロップダウンリストからお使いの Doris Spark Connector と一致するエンジンバージョンを選択し、Normal Network Connection にはステップ 2 で作成した接続を選択し、Spark Configuration には Doris Spark Connector をロードするために次のパラメーターを追加します。

   spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar

   oss://<bucketname>/path/connector.jar を、手順 1 でアップロードした Doris Spark コネクタ JAR ファイルの OSS パスに置き換えます。例： oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar

2. Development ページで SparkSQL タスクを作成し、右上隅から作成した SQL セッションを選択します。

   詳細については、「SparkSQLの開発」をご参照ください。

3. 次のコードを新しい SparkSQL タブにコピーし、必要に応じてパラメーターを更新後、実行 をクリックします。

   CREATE TEMPORARY VIEW test
   USING doris
   OPTIONS(
     "table.identifier" = "testdb.test",
     "fenodes" = "<doris_address>:<http_port>",
     "user" = "<user>",
     "password" = "<password>"
   );
   SELECT * FROM test;

   各パラメーターの説明は次のとおりです。

   パラメーター	説明	例
   table.identifier	Doris のテーブル識別子。	別の Doris クラスターを使用する場合は、それに応じてパラメーターを設定してください。 EMR on ECS 上に作成されたクラスターを使用する場合は、次の値を使用します。 table.identifier: この例では testdb.test を使用します。 <doris_address>：EMR on ECS コンソールで、Doris クラスターのNodes ページに移動し、emr-master の横にあるアイコンをクリックして、プライベート IP アドレスを確認します。 <http_port>: デフォルトは 8031 です。 <user>: デフォルトのユーザー名は root です。 <password>: デフォルトのパスワードは空です。
   <doris_address>	Doris サービスを実行しているノードのプライベート IP アドレス。
   <http_port>	Doris が HTTP リクエストをリッスンするために使用するポート番号。
   <user>	Doris への接続に使用するユーザー名。
   <password>	Doris への接続に使用するパスワード。

   データが正常に返された場合、設定は正しいです。

ノートブックセッションを使用した Doris テーブルの読み込み

1. ノートブックセッションを作成します。詳細については、「ノートブックセッションの管理」をご参照ください。

   セッションを作成するときは、Engine Version ドロップダウンリストから、ご使用の Doris Spark Connector と一致するエンジンバージョンを選択します。Normal Network Connection で、ステップ 2 で作成した接続を選択します。Spark Configuration で、次のパラメーターを追加して Doris Spark Connector をロードします。

   spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar

   oss://<bucketname>/path/connector.jar を、手順 1 でアップロードした Doris Spark コネクタ JAR ファイルの OSS パスに置き換えます。例： oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar

2. Development ページで、タイプ [インタラクティブ開発] > [Notebook] のタスクを作成し、右上隅で作成した Notebook セッションを選択します。

   詳細については、「ノートブックセッションの管理」をご参照ください。

3. 新しい Notebook タブに次のコードをコピーし、必要に応じてパラメーターを更新して、実行 をクリックします。

   dorisSparkDF = spark.read.format("doris") \
     .option("doris.table.identifier", "testdb.test") \
     .option("doris.fenodes", "<doris_address>:<http_port>") \
     .option("user", "<user>") \
     .option("password", "<password>") \
     .load()
   dorisSparkDF.show(3)

   各パラメーターの説明は次のとおりです。

   パラメーター	説明	例
   table.identifier	Doris のテーブル識別子。	別の Doris クラスターを使用する場合は、それに応じてパラメーターを設定してください。 EMR on ECS 上に作成されたクラスターを使用する場合は、次の値を使用します。 table.identifier: この例では testdb.test を使用します。 <doris_address>：EMR on ECS コンソールで、Doris クラスターのNodes ページに移動し、emr-master の横にあるアイコンをクリックして、プライベート IP アドレスを確認します。 <http_port>: デフォルトは 8031 です。 <user>: デフォルトのユーザー名は root です。 <password>: デフォルトのパスワードは空です。
   <doris_address>	Doris サービスを実行しているノードのプライベート IP アドレス。
   <http_port>	Doris が HTTP リクエストをリッスンするために使用するポート番号。
   <user>	Doris への接続に使用するユーザー名。
   <password>	Doris への接続に使用するパスワード。

   testdb.test テーブルの最初の 3 行が返されます。データが正常に返された場合、設定は正しいです。

SQL セッションを使用した Doris テーブルへの書き込み

前の手順で作成した SparkSQL タブに次のコードをコピーし、必要に応じてパラメーターを更新して、実行 をクリックします。

CREATE TEMPORARY VIEW test_write
USING doris
OPTIONS(
  "table.identifier" = "testdb.test",
  "fenodes" = "<doris_address>:<http_port>",
  "user" = "<user>",
  "password" = "<password>"
);
INSERT INTO test_write VALUES (4, 'd'), (5, 'e');
SELECT * FROM test_write;

次のデータが表示された場合、書き込み操作は成功です。

期待される結果には、id と name の 2 つの列があり、5 行のデータが含まれます。行 (1, a)、(2, b)、(3, c) は元の行です。行 (4, d)、(5, e) は新しく挿入された行です。

ノートブックセッションを使用した Doris テーブルへの書き込み

前の手順で作成した Notebook タブに次のコードをコピーし、必要に応じてパラメーターを更新して、実行 をクリックします。

data = [(7, 'f'), (8, 'g')]
mockDataDF = spark.createDataFrame(data, ["id", "name"])
mockDataDF.write.mode("append").format("doris") \
  .option("doris.table.identifier", "testdb.test") \
  .option("doris.fenodes", "<doris_address>:<http_port>") \
  .option("user", "<user>") \
  .option("password", "<password>") \
  .save()
dorisSparkDF = spark.read.format("doris") \
  .option("doris.table.identifier", "testdb.test") \
  .option("doris.fenodes", "<doris_address>:<http_port>") \
  .option("user", "<user>") \
  .option("password", "<password>") \
  .load()
dorisSparkDF.show(10)

次のデータが返された場合、書き込み操作は成功です。

+---+-----+
| id| name|
+---+-----+
|  1|    a|
|  2|    b|
|  3|    c|
|  7|    f|
|  8|    g|
+---+-----+