Realtime Compute for Apache Flink：Java

Langkah-langkah Pencegahan

• UDF dapat menyebabkan konflik dependensi dalam file JAR yang dikemas. Perhatikan hal-hal berikut saat mengembangkan UDF:

  • Pastikan versi Realtime Compute for Apache Flink yang Anda konfigurasikan untuk penyebaran SQL sesuai dengan versi Apache Flink yang Anda tentukan dalam file pom.xml.

  • Tentukan <scope>provided</scope> untuk dependensi terkait Apache Flink.

  • Gunakan Plugin Shade untuk mengemas dependensi pihak ketiga lainnya. Untuk informasi lebih lanjut, lihat Apache Maven Shade Plugin.

  Untuk informasi tentang cara menyelesaikan konflik dependensi, lihat FAQ tentang manajemen dan operasi ruang kerja serta namespace.

• Pemanggilan UDF secara sering dalam penyebaran SQL dapat menyebabkan timeout. Kami merekomendasikan agar Anda mengunggah file JAR dari UDF sebagai dependensi tambahan dari penyebaran dan menggunakan pernyataan CREATE TEMPORARY FUNCTION untuk mendeklarasikan UDF. Contoh: CREATE TEMPORARY FUNCTION 'GetJson' AS 'com.soybean.function.udf.MyGetJsonObjectFunction';.

Tipe UDF

Daftarkan UDF

• Untuk informasi tentang cara mendaftarkan UDF secara global, lihat Daftarkan catalog UDF.

• Untuk informasi tentang cara mendaftarkan UDF untuk penyebaran tunggal, lihat Daftarkan UDF tingkat penyebaran.

Pass parameter UDF

Anda dapat mengonfigurasi parameter UDF di konsol pengembangan Realtime Compute for Apache Flink dan mendapatkan nilai parameter di dalam UDF.

Parameter bernama

Dalam SQL, nilai parameter harus dilewatkan ke fungsi sesuai dengan urutan parameter yang didefinisikan dan parameter opsional tidak dapat dihilangkan. Jika jumlah parameter besar, memastikan urutan parameter yang benar bisa menjadi tantangan. Untuk memastikan urutan parameter yang benar dan meningkatkan kemudahan penggunaan, Anda dapat menggunakan parameter bernama untuk menentukan hanya parameter yang ingin Anda gunakan. Contoh berikut menunjukkan cara menggunakan parameter bernama. Dalam contoh ini, UDSF bernama ScalarFunction digunakan.

// Implementasikan UDSF di mana parameter kedua dan ketiga 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 Anda memanggil fungsi dalam SQL, Anda dapat menentukan hanya parameter pertama yang wajib. Anda juga dapat menentukan parameter kedua dan ketiga yang opsional. Contoh kode:

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 kedua yang opsional.
    MyNamedUdf(f1 => c, f2 => a) arg2_res,
    -- Tentukan parameter pertama yang wajib dan parameter ketiga yang opsional.
    MyNamedUdf(f1 => c, f3 => d) arg3_res
FROM s1;

Referensi

• Untuk informasi tentang cara menggunakan UDAF untuk mengurutkan dan menggabungkan data, lihat Gunakan UDAF untuk mengurutkan dan menggabungkan data.

• Untuk informasi tentang cara mengembangkan dan menggunakan Java UDF, lihat UDAFs, UDSFs, dan UDTFs.

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

• Untuk informasi tentang cara mengembangkan dan menggunakan Python UDF, lihat UDAFs, UDSFs, dan UDTFs.