E-MapReduce:Read from and write to Doris

Read Doris tables using an SQL session

1. Create an SQL session. For more information, see Manage SQL sessions.

   When creating the session, select the engine version that matches your Doris Spark Connector from the Engine Version drop-down list. For Normal Network Connection, select the connection created in Step 2. For Spark Configuration, add the following parameter to load the Doris Spark Connector.

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

   Replace oss://<bucketname>/path/connector.jar with the OSS path of the Doris Spark Connector JAR file that you uploaded in Step 1. For example: oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar.

2. On the Development page, create a SparkSQL task, and then select the SQL session that you created from the upper-right corner.

   For more information, see Develop SparkSQL.

3. You can copy the following code into the new SparkSQL tab, update the parameters as needed, and then click Run.

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

   The following describes the parameters.

   Parameter	Description	Example
   testdb.test	The actual database and table name in Doris.	If you use a different Doris cluster, configure the parameters accordingly. If you use a cluster created on EMR on ECS, use the following values: testdb.test: This example uses testdb.test. <doris_address>: In the EMR on ECS console, go to the Doris cluster’s Nodes page, click the icon next to emr-master, and view the private IP address. <http_port>: Default is 8031. <user>: Default username is root. <password>: Default password is empty.
   <doris_address>	The private IP address of the node running the Doris service.
   <http_port>	The port number Doris uses to listen for HTTP requests.
   <user>	The username used to connect to Doris.
   <password>	The password used to connect to Doris.

   If data is returned successfully, your configuration is correct.

Read Doris tables using a Notebook session

1. Create a Notebook session. For more information, see Manage Notebook sessions.

   When creating the session, select the engine version that matches your Doris Spark Connector from the Engine Version drop-down list. Under Normal Network Connection, select the connection that you created in Step 2. Under Spark Configuration, add the following parameter to load the Doris Spark Connector.

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

   Replace oss://<bucketname>/path/connector.jar with the OSS path of the Doris Spark Connector JAR file that you uploaded in Step 1. For example: oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar.

2. On the Development page, create a task of type Interactive development > Notebook, and then select the Notebook session that you created in the upper-right corner.

   For more information, see Manage Notebook sessions.

3. Copy the following code into the new Notebook tab, update the parameters as needed, and click Run.

   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)

   The following describes the parameters.

   Parameter	Description	Example
   testdb.test	The actual database and table name in Doris.	If you use a different Doris cluster, configure the parameters accordingly. If you use a cluster created on EMR on ECS, use the following values: testdb.test: This example uses testdb.test. <doris_address>: In the EMR on ECS console, go to the Doris cluster’s Nodes page, click the icon next to emr-master, and view the private IP address. <http_port>: Default is 8031. <user>: Default username is root. <password>: Default password is empty.
   <doris_address>	The private IP address of the node running the Doris service.
   <http_port>	The port number Doris uses to listen for HTTP requests.
   <user>	The username used to connect to Doris.
   <password>	The password used to connect to Doris.

   The first three rows of the testdb.test table are returned. If data is returned successfully, your configuration is correct.

Write to Doris tables using an SQL session

Copy the following code into the SparkSQL tab created in the previous step, update the parameters as needed, and click Run.

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;

If the following data is displayed, the write operation is successful.

The expected result contains two columns, id and name, with five rows. The rows (1, a), (2, b), and (3, c) are the original rows. The rows (4, d) and (5, e) are the newly inserted rows.

Write to Doris tables using a Notebook session

Copy the following code into the Notebook tab created in the previous step, update the parameters as needed, and click Run.

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)

If the following data is returned, the write operation was successful.

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