Realtime Compute for Apache Flink:Java

Flink aşağıdaki üç UDF türünü destekler.

UDF'ler isteğe bağlı bir open(FunctionContext context) yöntemi sağlar. FunctionContext, özel yapılandırma öğelerini parametre olarak aktarır. Süreç şu şekildedir:

1. Flink geliştirme konsolundaki O&M > Deployments sayfasının Configuration sekmesinde, Parameters bölümünün Other Configuration kısmına pipeline.global-job-parameters yapılandırma öğesini ekleyin.

   pipeline.global-job-parameters: | 
     'k1:{hi,hello}',
     'k2:"str:ing,str:ing"',
     'k3:"str""ing,str:ing"'

   FunctionContext#getJobParameter yalnızca pipeline.global-job-parameters içinde tanımlanan değerleri alabilir. UDF'nin kullandığı tüm yapılandırma öğelerini bu parametreye dahil etmelisiniz. Aşağıdaki adımlarda bu öğenin nasıl yapılandırılacağı açıklanmaktadır.

   Adım	Eylem	Yordam	Örnek
   1. Adım	Anahtar-değer çiftlerini tanımlayın.	Anahtarı ve değeri iki nokta üst üste (:) ile ayırın ve her anahtar-değer çiftini tek tırnak içine (') alın. Not Bir anahtar veya değer iki nokta üst üste (:) içeriyorsa anahtarı veya değeri çift tırnak içine (") alın. Bir anahtar veya değer iki nokta üst üste (:) veya çift tırnak (") içeriyorsa bunu ardışık iki çift tırnak ("") ile kaçış karakteri olarak belirtmeniz gerekir.	Anahtar = k1 ve değer = {hi,hello} ise çifti 'k1:{hi,hello}' olarak tanımlayın. Anahtar = k2 ve değer = str:ing,str:ing ise çifti 'k2:"str:ing,str:ing"' olarak tanımlayın. Anahtar = k3 ve değer = str"ing,str:ing ise çifti 'k3:"str""ing,str:ing"' olarak tanımlayın.
   2. Adım	Nihai pipeline.global-job-parameters değerini bir YAML dosyası olarak biçimlendirin.	Her anahtar-değer çiftini yeni bir satıra yerleştirin ve virgül (,) ile ayırın. Not YAML dosyasındaki çok satırlı bir dize dikey çubuk (|) ile başlar. YAML dosyasındaki çok satırlı bir dizenin her satırı aynı girintiye sahip olmalıdır.	pipeline.global-job-parameters: | 'k1:{hi,hello}', 'k2:"str:ing,str:ing"', 'k3:"str""ing,str:ing"'

2. UDF kodunda, her öğenin değerini almak için FunctionContext#getJobParameter kullanın. Aşağıdaki kod bir örnektir.

   Örnek:

   context.getJobParameter("k1", null); // Returns the string {hi,hello}.
   context.getJobParameter("k2", null); // Returns the string str:ing,str:ing.
   context.getJobParameter("k3", null); // Returns the string str"ing,str:ing.
   context.getJobParameter("pipeline.global-job-parameters", null); // Returns null. You can only get the content defined in pipeline.global-job-parameters, not any other job configuration item.

// Implement a user-defined scalar function. The last two input parameters are optional (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;
	}
}

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,
    -- Specify only the first required parameter
    MyNamedUdf(f1 => c) arg1_res,
    -- Specify the first required parameter and the second optional parameter
    MyNamedUdf(f1 => c, f2 => a) arg2_res,
    -- Specify the first required parameter and the third optional parameter
    MyNamedUdf(f1 => c, f3 => d) arg3_res
FROM s1;