Baca dari dan tulis ke Doris - E-MapReduce - Alibaba Cloud Documentation Center

Anda dapat menghubungkan EMR Serverless Spark ke Doris menggunakan Apache Doris Spark Connector resmi dan menambahkan konfigurasi yang diperlukan. Topik ini menjelaskan cara membaca data dari dan menulis data ke Doris dalam lingkungan EMR Serverless Spark.

Informasi latar belakang

Apache Doris adalah database analitik real-time berkinerja tinggi yang mendukung skenario seperti analisis laporan, kueri ad hoc, dan akselerasi query terfederasi pada data lake. Untuk informasi selengkapnya, lihat Pengenalan Apache Doris.

EMR Serverless Spark adalah Produk Lakehouse berkinerja tinggi yang kompatibel dengan Spark open source dan menyediakan layanan platform data enterprise-grade yang sepenuhnya dikelola. Dengan menggabungkan Apache Doris dan EMR Serverless Spark, Anda dapat membaca, menulis, dan menganalisis data secara efisien untuk mengimplementasikan alur kerja pemrosesan data end-to-end.

Prasyarat

Ruang kerja Serverless Spark telah dibuat. Untuk informasi selengkapnya, lihat Buat ruang kerja.
Kluster Doris telah dibuat.

Jika Anda membuat kluster OLAP EMR on ECS yang mencakup layanan Doris, lihat Buat kluster. Topik ini menggunakan contoh kluster Doris yang dibuat di EMR on ECS, yang selanjutnya disebut sebagai kluster EMR Doris.

Batasan

Versi engine EMR Serverless Spark harus esr-2.6.0, esr-3.2.0, esr-4.2.0, atau versi yang lebih baru.

Prosedur

Langkah 1: Dapatkan JAR Doris Spark Connector dan unggah ke OSS

Tinjau dokumentasi resmi Apache Doris untuk Spark Doris Connector. Dokumentasi tersebut mencantumkan kompatibilitas antara versi connector dan versi engine Spark. Pastikan versi Spark Anda kompatibel dengan versi Doris Spark Connector.

Buka repositori GitHub Doris Spark Connector dan unduh versi yang sesuai.

File Java Archive (JAR) Doris Spark Connector mengikuti format penamaan berikut: spark-doris-connector-spark-${spark_version}-${connector_version}.jar. Misalnya, jika versi engine Anda adalah esr-3.1.0 (Spark 3.4.3, Scala 2.12), unduh spark-doris-connector-spark-3.4-24.0.0.jar.
Unggah file JAR Spark Connector yang telah diunduh ke Alibaba Cloud OSS. Untuk informasi selengkapnya, lihat Simple upload.

Langkah 2: Buat koneksi jaringan

EMR Serverless Spark harus membuat koneksi jaringan dengan kluster EMR Doris agar dapat mengakses layanan Doris. Untuk informasi selengkapnya tentang koneksi jaringan, lihat Koneksi jaringan antara EMR Serverless Spark dan VPC lainnya.

Penting

Saat menambahkan aturan security group, buka hanya Port yang diperlukan di Destination Port Range. Nilai port yang valid berkisar dari 1 hingga 65535. Contohnya, buka port HTTP (8031), port RPC (9061), dan port Webserver (8041).

Langkah 3: Buat database dan tabel di kluster EMR Doris

Login ke kluster menggunakan SSH. Untuk informasi selengkapnya, lihat Login ke kluster.
Jalankan perintah berikut untuk menghubungkan ke kluster EMR Doris.
```
mysql -h127.0.0.1  -P 9031 -uroot
```

Buat database dan tabel.

CREATE DATABASE IF NOT EXISTS testdb;
USE testdb;
CREATE TABLE test (
    id INT, 
    name STRING
) PROPERTIES("replication_num" = "1");

Masukkan data uji.

INSERT INTO test VALUES (1, 'a'), (2, 'b'), (3, 'c');

Kueri data tersebut.

SELECT * FROM test;

Output berikut dikembalikan.

MySQL [testdb]> SELECT * FROM test;
+------+------+
| id   | name |
+------+------+
|    1 | a    |
|    2 | b    |
|    3 | c    |
+------+------+
3 rows in set (0.177 sec)

Langkah 4: Baca tabel Doris menggunakan Serverless Spark

Baca tabel Doris menggunakan session SQL

Buat session SQL. Untuk informasi selengkapnya, lihat Kelola session SQL.
Saat membuat session, pilih versi engine yang sesuai dengan Doris Spark Connector Anda dari daftar drop-down Engine Version. Untuk Normal Network Connection, pilih koneksi yang dibuat di Langkah 2. Untuk Spark Configuration, tambahkan parameter berikut untuk memuat Doris Spark Connector.
```
spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar
```
Ganti oss://<bucketname>/path/connector.jar dengan path OSS file JAR Doris Spark Connector yang diunggah di Langkah 1. Contohnya: oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar.
Di halaman Development, buat task SparkSQL, lalu pilih session SQL yang telah dibuat dari pojok kanan atas.

Untuk informasi selengkapnya, lihat Kembangkan SparkSQL.

Anda dapat menyalin kode berikut ke tab SparkSQL baru, sesuaikan parameter sesuai kebutuhan, lalu klik Run.

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

Berikut ini penjelasan parameter-parameter tersebut.

Parameter	Deskripsi	Contoh
`testdb.test`	Nama database dan tabel aktual di Doris.	Jika Anda menggunakan kluster Doris yang berbeda, konfigurasikan parameter sesuai kebutuhan. Jika Anda menggunakan kluster yang dibuat di EMR on ECS, gunakan nilai-nilai berikut: `testdb.test`: Contoh ini menggunakan `testdb.test`. `<doris_address>`: Di Konsol EMR on ECS, buka halaman Nodes kluster Doris, klik ikon di samping emr-master, lalu lihat Alamat IP pribadi. `<http_port>`: Default-nya adalah 8031. `<user>`: Username default adalah `root`. `<password>`: Password default kosong.
`<doris_address>`	Alamat IP pribadi node yang menjalankan layanan Doris.
`<http_port>`	Nomor port yang digunakan Doris untuk menerima permintaan HTTP.
`<user>`	Username yang digunakan untuk menghubungkan ke Doris.
`<password>`	Password yang digunakan untuk menghubungkan ke Doris.

Jika data berhasil dikembalikan, konfigurasi Anda sudah benar.

Baca tabel Doris menggunakan session Notebook

Buat session Notebook. Untuk informasi selengkapnya, lihat Kelola session Notebook.
Saat membuat session, pilih versi engine yang sesuai dengan Doris Spark Connector Anda dari daftar drop-down Engine Version. Di bawah Normal Network Connection, pilih koneksi yang dibuat di Langkah 2. Di bawah Spark Configuration, tambahkan parameter berikut untuk memuat Doris Spark Connector.
```
spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar
```
Ganti oss://<bucketname>/path/connector.jar dengan path OSS file JAR Doris Spark Connector yang diunggah di Langkah 1. Contohnya: oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar.
Di halaman Development, buat task bertipe Interactive development > Notebook, lalu pilih session Notebook yang telah dibuat di pojok kanan atas.

Untuk informasi selengkapnya, lihat Kelola session Notebook.

Salin kode berikut ke tab Notebook baru, sesuaikan parameter sesuai kebutuhan, lalu klik 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)

Berikut ini penjelasan parameter-parameter tersebut.

Parameter	Deskripsi	Contoh
`testdb.test`	Nama database dan tabel aktual di Doris.	Jika Anda menggunakan kluster Doris yang berbeda, konfigurasikan parameter sesuai kebutuhan. Jika Anda menggunakan kluster yang dibuat di EMR on ECS, gunakan nilai-nilai berikut: `testdb.test`: Contoh ini menggunakan `testdb.test`. `<doris_address>`: Di Konsol EMR on ECS, buka halaman Nodes kluster Doris, klik ikon di samping emr-master, lalu lihat Alamat IP pribadi. `<http_port>`: Default-nya adalah 8031. `<user>`: Username default adalah `root`. `<password>`: Password default kosong.
`<doris_address>`	Alamat IP pribadi node yang menjalankan layanan Doris.
`<http_port>`	Nomor port yang digunakan Doris untuk menerima permintaan HTTP.
`<user>`	Username yang digunakan untuk menghubungkan ke Doris.
`<password>`	Password yang digunakan untuk menghubungkan ke Doris.

Tiga baris pertama dari tabel testdb.test dikembalikan. Jika data berhasil dikembalikan, konfigurasi Anda sudah benar.

Langkah 5: Tulis ke tabel Doris menggunakan Serverless Spark

Tulis ke tabel Doris menggunakan session SQL

Salin kode berikut ke tab SparkSQL yang dibuat pada langkah sebelumnya, sesuaikan parameter sesuai kebutuhan, lalu klik 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;

Jika data berikut ditampilkan, operasi penulisan berhasil.

Hasil yang diharapkan berisi dua kolom, id dan name, dengan lima baris. Baris (1, a), (2, b), dan (3, c) adalah baris asli. Baris (4, d) dan (5, e) adalah baris yang baru dimasukkan.

Tulis ke tabel Doris menggunakan session Notebook

Salin kode berikut ke tab Notebook yang dibuat pada langkah sebelumnya, sesuaikan parameter sesuai kebutuhan, lalu klik 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)

Jika data berikut dikembalikan, operasi penulisan berhasil.

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