Realtime Compute for Apache Flink:Java

Catatan penting

• Untuk menghindari konflik dependensi paket JAR saat mengembangkan UDF, pertimbangkan hal berikut:

  • Pastikan versi Flink yang dipilih pada halaman pengembangan SQL sesuai dengan versi Flink dalam dependensi Pom.

  • Untuk dependensi terkait Flink, atur cakupannya ke `provided` dengan menambahkan <scope>provided</scope>.

  • Paketkan dependensi pihak ketiga lainnya menggunakan metode Shade. Untuk informasi selengkapnya, lihat Apache Maven Shade Plugin.

  Untuk informasi lebih lanjut tentang konflik dependensi Flink, lihat Bagaimana cara menyelesaikan konflik dependensi Flink?.

• Untuk mencegah timeout akibat pemanggilan UDF yang terlalu sering dalam pekerjaan SQL, unggah paket JAR UDF sebagai file dependensi, lalu deklarasikan fungsi tersebut dalam pekerjaan menggunakan sintaksis CREATE TEMPORARY FUNCTION. Contohnya: CREATE TEMPORARY FUNCTION 'GetJson' AS 'com.soybean.function.udf.MyGetJsonObjectFunction';

Klasifikasi UDF

Pendaftaran UDF

• Untuk informasi tentang cara mendaftarkan UDF global, lihat Global UDFs.

• Untuk informasi tentang cara mendaftarkan UDF tingkat pekerjaan, lihat Job-level UDFs.

Pass parameter ke UDF

Anda dapat mengonfigurasi parameter untuk UDF di Konsol pengembangan Flink dan menggunakannya dalam kode UDF. Hal ini memungkinkan Anda mengubah nilai parameter UDF secara langsung di konsol dengan cepat.

Parameter bernama

Saat memanggil fungsi dalam SQL, Anda harus menentukan semua parameter dalam urutan yang benar. Jika suatu fungsi memiliki banyak parameter, sangat mudah membuat kesalahan dalam jumlah atau urutan parameternya. Anda juga tidak dapat mengabaikan parameter opsional. Parameter bernama memungkinkan Anda hanya menentukan parameter yang diperlukan, sehingga mengurangi kemungkinan kesalahan dan lebih nyaman. Contoh berikut dari user-defined scalar function (ScalarFunction) menunjukkan cara menggunakan parameter bernama.

// Implementasikan user-defined scalar function. Dua parameter input terakhir bersifat opsional (isOptional = true).
public class MyFuncWithNamedArgs extends ScalarFunction  {
	private static final long serialVersionUID = 1L;

	public String eval(@ArgumentHint(name = "f1", isOptional = false, type = @DataTypeHint("STRING")) String f1,
			@ArgumentHint(name = "f2", isOptional = true, type = @DataTypeHint("INT")) Integer i2,
			@ArgumentHint(name = "f3", isOptional = true, type = @DataTypeHint("LONG")) Long l3) {

		if (i2 != null) {
			return "i2#" + i2;
		}
		if (l3 != null) {
			return "l3#" + l3;
		}
		return "default#" + f1;
	}
}

Saat menggunakan UDF ini dalam SQL, Anda dapat hanya menentukan parameter pertama yang wajib atau secara selektif menentukan parameter opsional. Kode berikut merupakan contohnya.

CREATE TEMPORARY FUNCTION MyNamedUdf AS 'com.aliyun.example.MyFuncWithNamedArgs';

CREATE temporary TABLE s1 (
    a INT,
    b BIGINT,
    c VARCHAR,
    d VARCHAR,
    PRIMARY KEY(a) NOT ENFORCED
) WITH (
    'connector' = 'datagen',
    'rows-per-second'='1'
);

CREATE temporary TABLE sink (
    a INT,
    b VARCHAR,
    c VARCHAR,
    d VARCHAR
) WITH (
    'connector' = 'print'
);

INSERT INTO sink
SELECT a,
    -- Tentukan hanya parameter pertama yang wajib
    MyNamedUdf(f1 => c) arg1_res,
    -- Tentukan parameter pertama yang wajib dan parameter opsional kedua
    MyNamedUdf(f1 => c, f2 => a) arg2_res,
    -- Tentukan parameter pertama yang wajib dan parameter opsional ketiga
    MyNamedUdf(f1 => c, f3 => d) arg3_res
FROM s1;

Referensi

• Gunakan UDAF untuk mengurutkan dan mengagregasi data.

• Untuk contoh cara mengembangkan dan menggunakan UDF Java, lihat User-defined aggregate functions (UDAFs), User-defined scalar functions (UDSFs), dan User-defined table-valued functions (UDTFs).

• Untuk informasi tentang cara men-debug dan menyetel UDF Python, lihat Ikhtisar.

• Untuk contoh cara mengembangkan dan menggunakan UDF Python, lihat User-defined aggregate functions (UDAFs), User-defined scalar functions (UDSFs), dan User-defined table-valued functions (UDTFs).