Realtime Compute for Apache Flink:Pekerjaan dialek Hive

Prasyarat

• Jika Anda mengakses Konsol sebagai pengguna RAM, peran RAM, atau identitas lainnya, pastikan Anda memiliki izin yang diperlukan. Untuk informasi selengkapnya, lihat izin.

• Anda telah membuat ruang kerja. Untuk informasi selengkapnya, lihat Aktifkan Realtime Compute for Apache Flink.

Batasan

• Hanya Ververica Runtime (VVR) 8.0.11 dan versi yang lebih baru yang mendukung dialek Hive.

• Pekerjaan SQL saat ini hanya mendukung sintaks pernyataan INSERT dari dialek Hive, dan Anda harus mendeklarasikan USE Catalog <yourHiveCatalog> sebelum pernyataan INSERT. Jika Anda perlu membuat tabel, lakukan operasi tersebut di halaman Scripts.

• Fungsi user-defined (UDF) Hive dan Flink tidak didukung.

Langkah 1: Buat katalog Hive

1. Konfigurasikan metadata Hive. Untuk informasi selengkapnya, lihat Konfigurasikan metadata Hive.

2. Buat katalog Hive. Untuk informasi selengkapnya, lihat Buat katalog Hive.

   Dalam tutorial ini, katalog Hive diberi nama hdfshive.

Langkah 2: Siapkan tabel Hive sampel

1. Di panel navigasi sebelah kiri, buka Development > Scripts. Klik New untuk membuat skrip.

2. Jalankan pernyataan SQL sampel berikut.

   Penting

   Tabel sumber Hive dan tabel sink harus berupa tabel permanen yang dibuat dengan pernyataan CREATE TABLE. Anda tidak dapat menggunakan tabel temporary yang dibuat dengan pernyataan CREATE TEMPORARY TABLE.

   -- Gunakan katalog Hive. Dalam contoh ini, katalog bernama hdfshive dan dibuat pada Langkah 1.
   USE CATALOG hdfshive;   
   -- Buat tabel sumber dengan format penyimpanan default.
   CREATE TABLE source_table (
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )WITH ('connector' = 'hive');
   -- Buat tabel sink dengan format penyimpanan default.
   CREATE TABLE target_table (
   city STRING,
   avg_salary FLOAT,
   user_count INT
   )WITH ('connector' = 'hive');
   -- Masukkan data sampel ke dalam tabel sumber.
   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);
   -- Buat tabel dengan format penyimpanan tertentu, misalnya Parquet.
   -- Muat modul Hive.
   load MODULE hive with ('hive-version' = '2.3.6');
   use CATALOG `hdfshive`;
   -- Wajib: Atur dialek SQL ke 'hive' agar mengenali kata kunci DDL Hive seperti 'STORED'.
   set 'table.sql-dialect' = 'hive';
   CREATE TABLE `parquet_table`(
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )STORED AS PARQUET;

Langkah 3: Buat pekerjaan Hive SQL

1. Di panel navigasi sebelah kiri, buka Development > ETL.

2. Klik New. Di kotak dialog New Draft, pilih Blank Batch Draft (BETA) lalu klik Next.

3. Masukkan informasi pekerjaan.

   Parameter	Deskripsi	Contoh
   Name	Nama pekerjaan. Catatan Nama pekerjaan harus unik dalam ruang kerja saat ini.	hive-sql
   Location	Folder tempat file kode pekerjaan disimpan. Anda juga dapat mengklik ikon di sebelah kanan folder yang sudah ada untuk membuat subfolder.	Drafts
   Engine version	Versi engine Flink yang digunakan oleh pekerjaan. Kami menyarankan Anda memilih versi dengan tag RECOMMENDED. Versi ini menawarkan keandalan dan performa yang lebih tinggi. Untuk informasi selengkapnya tentang versi engine, lihat Catatan rilis dan Versi engine.	vvr-8.0.11-flink-1.17
   SQL dialect	Bahasa SQL untuk pemrosesan data. Catatan Parameter ini hanya muncul jika Anda memilih versi engine yang mendukung dialek Hive.	Hive SQL

4. Klik Create.

Langkah 4: Tulis dan terapkan pekerjaan Hive SQL

1. Tulis pernyataan SQL.

   Contoh ini menghitung jumlah pengguna yang berusia di atas 30 tahun dan gaji rata-rata untuk setiap kota. Anda dapat menyalin skrip SQL berikut ke editor SQL.

   -- Gunakan katalog Hive. Dalam contoh ini, katalog bernama hdfshive dan dibuat pada Langkah 1.
   USE CATALOG hdfshive; 
   INSERT INTO TABLE target_table
   SELECT
     city,
     AVG(salary) AS avg_salary, -- Hitung gaji rata-rata
     COUNT(id) AS user_count -- Hitung jumlah pengguna
   FROM source_table
   WHERE age > 30 -- Filter pengguna yang berusia di atas 30
   GROUP BY city; -- Kelompokkan berdasarkan kota

2. Di pojok kanan atas, klik Deploy. Di kotak dialog, konfigurasikan parameter sesuai kebutuhan (tutorial ini menggunakan pengaturan default) lalu klik OK.

(Opsional) Langkah 5: Konfigurasikan parameter waktu proses

1. Di panel navigasi sebelah kiri, buka O&M > Deployments.

2. Dari daftar drop-down, pilih Batch job. Temukan pekerjaan target lalu klik Details di kolom Actions.

3. Di panel detail penerapan, klik Edit pada bagian Runtime parameters configuration.

4. Di bidang 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 selengkapnya tentang parameter ini, lihat Tulis data ke OSS-HDFS.

5. Klik Save.

Langkah 6: Mulai pekerjaan dan lihat hasilnya

1. Di halaman Deployments, pilih Batch job dari filter, temukan pekerjaan target Anda (misalnya, hive-sql), lalu klik Start di kolom Actions.

2. Setelah status pekerjaan berubah menjadi FINISHED, lihat hasilnya.

   Di halaman Development > Scripts, jalankan pernyataan SQL berikut untuk melihat data, yang mencakup jumlah pengguna berusia di atas 30 tahun dan gaji rata-rata mereka di setiap kota.

   -- Gunakan katalog Hive. Dalam contoh ini, katalog bernama hdfshive dan dibuat pada Langkah 1.
   USE CATALOG hdfshive; 
   select * from target_table;

   Kueri mengembalikan tiga baris dari target_table dengan kolom city, avg_salary, dan user_count: Los Angeles (9.000,0, 1), New York (7.000,0, 1), dan San Francisco (8.000,0, 1).

Pengembangan pekerjaan Hive JAR

Anda dapat menjalankan pekerjaan dialek Hive sebagai pekerjaan JAR. Hal ini memerlukan versi 11.2 atau yang lebih baru dari paket JAR "ververica-connector-hive-2.3.6". Anda juga harus memastikan bahwa konfigurasi Hive dalam pekerjaan JAR dan pengaturan Konsol sesuai.

1. Pengaturan Konsol

   1. URI JAR menentukan paket JAR yang diunggah untuk pekerjaan JAR.

   2. Di Additional Dependencies, unggah empat file konfigurasi dari kluster Hive Anda: core-site.xml, mapred-site.xml, hdfs-site.xml, dan hive-site.xml. Anda juga harus mengunggah paket JAR ververica-connector-hive-2.3.6.

   3. Konfigurasikan parameter waktu proses. Berdasarkan konfigurasi kluster Hive Anda, jika Anda perlu menulis data ke OSS-HDFS, gunakan pengaturan dari (Opsional) Langkah 5: Konfigurasikan parameter waktu proses.

      table.sql-dialect: HIVE
      classloader.parent-first-patterns.additional: org.apache.hadoop;org.antlr.runtime
      kubernetes.application-mode.classpath.include-user-jar: true

2. Kode contoh pekerjaan JAR:

   1. 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`;");

Dokumen terkait

• Untuk informasi selengkapnya tentang sintaks pernyataan INSERT untuk dialek Hive, lihat Pernyataan INSERT | Apache Flink.

• Untuk informasi selengkapnya tentang pemrosesan data batch dengan Flink SQL, lihat Pemrosesan batch.