Realtime Compute for Apache Flink:Java

Observações importantes

• Para evitar conflitos de dependência de pacotes JAR durante o desenvolvimento de UDFs, considere o seguinte:

  • Verifique se a versão do Flink selecionada na página de desenvolvimento SQL corresponde à versão do Flink na dependência do Pom.

  • Para dependências relacionadas ao Flink, defina o escopo como provided adicionando <scope>provided</scope>.

  • Empacote outras dependências de terceiros usando o método Shade. Para mais informações, consulte Apache Maven Shade Plugin.

  Para obter mais detalhes sobre conflitos de dependência do Flink, consulte How do I resolve Flink dependency conflicts?.

• Para evitar timeouts causados por chamadas frequentes a uma UDF em um job SQL, faça upload do pacote JAR da UDF como arquivo de dependência. Em seguida, declare a função no job usando a sintaxe CREATE TEMPORARY FUNCTION. Exemplo: CREATE TEMPORARY FUNCTION 'GetJson' AS 'com.soybean.function.udf.MyGetJsonObjectFunction';

Classificações de UDF

Registro de UDF

• Para saber como registrar uma UDF global, consulte Global UDFs.

• Para saber como registrar uma UDF no nível do job, consulte Job-level UDFs.

Passagem de parâmetros para uma UDF

Configure parâmetros para uma UDF no console de desenvolvimento do Flink e referencie-os no código da UDF para modificar rapidamente os valores diretamente no console.

Parâmetros nomeados

Chamadas de função padrão exigem todos os parâmetros na ordem correta, o que aumenta a probabilidade de erros em funções com muitos parâmetros e impede a omissão de parâmetros opcionais. Parâmetros nomeados permitem especificar apenas os necessários. O exemplo de ScalarFunction a seguir demonstra como utilizar parâmetros nomeados.

// 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;
	}
}

Ao usar essa UDF em SQL, especifique apenas o parâmetro obrigatório ou inclua seletivamente os parâmetros opcionais.

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;

Referências

• Use a UDAF to sort and aggregate data.

• Para exemplos de desenvolvimento e uso de UDFs Java, consulte User-defined aggregate functions (UDAFs), User-defined scalar functions (UDSFs) e User-defined table-valued functions (UDTFs).

• Para informações sobre depuração e ajuste de UDFs Python, consulte Overview.

• Para exemplos de desenvolvimento e uso de UDFs Python, consulte User-defined aggregate functions (UDAFs), User-defined scalar functions (UDSFs) e User-defined table-valued functions (UDTFs).