Memulai dengan penerapan Flink SQL - Realtime Compute for Apache Flink

Realtime Compute for Apache Flink mendukung dialek Hive dalam pengembangan pekerjaan SQL untuk pemrosesan batch, sehingga meningkatkan interoperabilitas dengan Apache Hive dan memungkinkan migrasi pekerjaan Apache Hive ke Realtime Compute for Apache Flink secara mulus.

Prasyarat

Pengguna RAM atau Peran RAM yang digunakan untuk mengakses konsol pengembangan Realtime Compute for Apache Flink harus memiliki izin yang diperlukan. Untuk informasi lebih lanjut, lihat Manajemen Izin.
Sebuah ruang kerja Realtime Compute for Apache Flink telah dibuat. Untuk informasi lebih lanjut, lihat Aktifkan Realtime Compute for Apache Flink.

Batasan

Hanya Realtime Compute for Apache Flink yang menggunakan Ververica Runtime (VVR) 8.0.11 atau versi lebih baru yang mendukung dialek Hive.
Hanya pernyataan INSERT dari dialek Hive yang didukung. Untuk menggunakan pernyataan INSERT, deklarasikan USE Catalog <yourHiveCatalog> sebelum pernyataan INSERT.
Fungsi yang ditentukan pengguna (UDF) Hive dan Flink tidak didukung.

Langkah 1: Buat katalog Hive

Konfigurasikan metadata Hive. Untuk informasi lebih lanjut, lihat Konfigurasikan Metadata Hive.
Buat katalog Hive. Untuk informasi lebih lanjut, lihat Buat Katalog Hive.
Dalam contoh ini, sebuah katalog Hive bernama hdfshive dibuat.

Langkah 2: Buat tabel Hive dan sisipkan data sampel

Di panel navigasi sisi kiri konsol pengembangan, pilih Development > Scripts. Pada halaman yang muncul, klik New untuk membuat skrip.

Eksekusi pernyataan SQL berikut.

Penting

Anda harus menggunakan tabel permanen yang dibuat dengan menjalankan pernyataan CREATE TABLE sebagai tabel sumber dan tabel sink Hive. Tabel sementara yang dibuat dengan menjalankan pernyataan CREATE TEMPORARY TABLE tidak dapat digunakan sebagai tabel sumber dan tabel sink.

-- Gunakan katalog hdfshive yang dibuat di Langkah 1.
USE CATALOG hdfshive;   

-- Buat tabel sumber.
CREATE TABLE source_table (
 id INT,
 name STRING,
 age INT,
 city STRING,
 salary FLOAT
)WITH ('connector' = 'hive');

-- Buat tabel hasil.
CREATE TABLE target_table (
city STRING,
avg_salary FLOAT,
user_count INT
)WITH ('connector' = 'hive');

-- Tulis data uji ke tabel hasil.
INSERT INTO source_table VALUES
(1, 'Alice', 25, 'New York', 5000.0),
(2, 'Bob', 30, 'San Francisco', 6000.0),
(3, 'Charlie', 35, 'New York', 7000.0),
(4, 'David', 40, 'San Francisco', 8000.0),
(5, 'Eva', 45, 'Los Angeles', 9000.0);

Langkah 3: Buat draf SQL Hive

Di panel navigasi sisi kiri konsol pengembangan Realtime Compute for Apache Flink, pilih Development > ETL.
Klik New. Dalam kotak dialog New Draft, pilih Blank Batch Draft dengan label BETA, lalu klik Next.

Masukkan informasi draf.

Parameter	Deskripsi	Contoh
Name	Nama draf yang ingin Anda buat. Catatan Nama draf harus unik di proyek saat ini.	hive-sql
Location	Folder tempat file kode draf disimpan. Anda juga dapat mengklik ikon di sebelah kanan folder yang ada untuk membuat subfolder.	Draft
Engine Version	Versi mesin draf yang akan dibuat. Kami merekomendasikan Anda menggunakan versi mesin dengan label RECOMMENDED. Versi dengan label ini memberikan keandalan dan performa yang lebih tinggi. Untuk informasi lebih lanjut tentang versi mesin, lihat Catatan Rilis dan Versi Mesin.	vvr-8.0.11-flink-1.17
SQL Dialect	Bahasa pemrosesan data SQL. Catatan Parameter ini hanya ditampilkan ketika versi mesin yang mendukung dialek Hive dipilih.	Hive SQL

Klik Create.

Langkah 4: Tulis kode dan terapkan draf SQL Hive

Masukkan pernyataan SQL di editor SQL.

Dalam contoh ini, jumlah pengguna yang usianya lebih dari 30 tahun dan rata-rata gaji pengguna di setiap kota dihitung. Anda dapat menyalin pernyataan SQL berikut ke editor SQL.

-- Gunakan katalog Hive yang dibuat di Langkah 1.
USE CATALOG hdfshive; 

INSERT INTO TABLE target_table
SELECT
  city,
  AVG(salary) AS avg_salary, -- Hitung rata-rata gaji pengguna.
  COUNT(id) AS user_count -- Hitung jumlah pengguna.
FROM source_table
WHERE age > 30 -- Filter pengguna yang usianya lebih dari 30.
GROUP BY city; -- Kelompokkan data berdasarkan kota.

Klik Deploy di pojok kanan atas. Di kotak dialog Terapkan draf, konfigurasikan parameter berdasarkan kebutuhan bisnis Anda. Dalam contoh ini, pertahankan nilai default. Lalu, klik Confirm.

(Opsional) Langkah 5: Konfigurasikan parameter untuk menjalankan penerapan

Penting

Jika Anda ingin mengakses kluster Hive menggunakan JindoSDK, Anda harus melakukan langkah ini.

Di panel navigasi sisi kiri konsol pengembangan Realtime Compute for Apache Flink, pilih O&M > Deployments.
Pilih BATCH dari daftar drop-down di bagian atas halaman Penerapan. Temukan penerapan yang ingin Anda kelola dan klik Details.
Pada bagian Parameters di panel yang muncul, klik Edit di pojok kanan atas.

Di bagian Other Configuration, tambahkan konfigurasi berikut:

fs.oss.jindo.endpoint: <YOUR_Endpoint> 
fs.oss.jindo.buckets: <YOUR_Buckets>
fs.oss.jindo.accessKeyId: <YOUR_AccessKeyId>
fs.oss.jindo.accessKeySecret: <YOUR_AccessKeySecret>

Untuk informasi lebih lanjut tentang parameter, lihat Tulis Data ke OSS-HDFS.

Di pojok kanan atas bagian tersebut, klik Save.

Langkah 6: Mulai penerapan SQL dan lihat hasil pemrosesan data

Temukan penerapan target dan klik Start.
Setelah penerapan memasuki status FINISHED, lihat hasil komputasi penerapan.
Pilih Development > Scripts. Pada halaman yang muncul, eksekusi pernyataan SQL sampel berikut untuk menanyakan jumlah pengguna yang usianya lebih dari 30 tahun dan rata-rata gaji mereka di setiap kota.
```
-- Gunakan katalog Hive yang dibuat di Langkah 1.
USE CATALOG hdfshive; 

select * from target_table;
```

Gunakan dialek Hive dengan Table API

Hologres VVR 11.2+ mendukung penggunaan dialek Hive dengan Table API. Untuk memastikan eksekusi pekerjaan yang benar, unduh JAR ververica-connector-hive-2.3.6 dan cermin konfigurasi Hive Anda dalam kode dan pengaturan pekerjaan Anda.

Buat penerapan JAR dan konfigurasikan pekerjaan.
1. Di halaman Deployments, buat penerapan JAR, lalu unggah JAR aplikasi Anda di JAR URI.
2. Untuk Additional Dependencies, unggah file konfigurasi kluster Hive (core-site.xml, mapred-site.xml, hdfs-site.xml, dan hive-site.xml) serta JAR ververica-connector-hive-2.3.6.
3. Buka halaman detail penerapan, di tab Configuration, pada bagian Parameters, masukkan parameter waktu proses pekerjaan Anda seperti berikut. Untuk menulis ke OSS-HDFS, lihat (Opsional) Langkah 5: Konfigurasi parameter untuk menjalankan penerapan.
```
table.sql-dialect: HIVE
classloader.parent-first-patterns.additional: org.apache.hadoop;org.antlr.runtime
kubernetes.application-mode.classpath.include-user-jar: true
```

Kode contoh:

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
Configuration conf = new Configuration();
conf.setString("type", "hive");
conf.setString("default-database", "default");
conf.setString("hive-version", "2.3.6"); 
conf.setString("hive-conf-dir", "/flink/usrlib/" );
conf.setString("hadoop-conf-dir", "/flink/usrlib/");

CatalogDescriptor descriptor = CatalogDescriptor.of("hivecat", conf);
tableEnv.createCatalog("hivecat", descriptor);

tableEnv.loadModule("hive", new HiveModule());
tableEnv.useModules("hive");
tableEnv.useCatalog("hivecat");

tableEnv.executeSql("insert into `hivecat`.`default`.`test_write` select * from `hivecat`.`default`.`test_read`;");

Referensi

Untuk informasi lebih lanjut tentang sintaks INSERT dari dialek Hive, lihat Pernyataan INSERT | Apache Flink.
Untuk informasi lebih lanjut tentang cara menggunakan fitur pemrosesan batch Realtime Compute for Apache Flink, lihat Memulai dengan Pemrosesan Batch Realtime Compute for Apache Flink.