Apache -Flink 1.11 Невозможно использовать Python UDF в SQL Function DDL - PullRequest
Новая
       106

Apache -Flink 1.11 Невозможно использовать Python UDF в SQL Function DDL

0 голосов
/ 09 июля 2020

Согласно этой странице слияния:

https://cwiki.apache.org/confluence/display/FLINK/FLIP-106%3A+Support+Python+UDF+in+SQL+Function+DDL

python udf во Flink 1.11 доступны для использования с функциями SQL.

I перешел к документации flink здесь:

https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/sqlClient.html

и попробуйте это на терминале и запустите sql -клиент. sh со следующими параметрами :

$ sql-client.sh embedded --pyExecutable /Users/jonathanfigueroa/opt/anaconda3/bin/python --pyFiles /Users/jonathanfigueroa/Desktop/pyflink/inference/test1.py

, а затем: 

> Create Temporary System Function func1 as 'test1.func1' Language PYTHON;
[INFO] Function has been created.

и когда я попробовал: 

> Select func1(str) From (VALUES ("Name1", "Name2", "Name3"));
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalStateException: Instantiating python function 'test1.func1' failed.

Я попытался использовать: -pyarch,--pyArchives, -pyexec,--pyExecutable, -pyfs,--pyFiles в каждой комбинации .zip, .py и всегда один и тот же результат.

кстати, мой python файл выглядит так: 

def func1(s):
    return s;

Что-то мне не хватает?

С уважением,

Джонатан

1 Ответ

0 голосов
/ 10 июля 2020

UDF python должен быть обернут декоратором udf в pyflink.table.udf, например: 

from pyflink.table.types import DataTypes
from pyflink.table.udf import udf

@udf(input_types=[DataTypes.INT()], result_type=DataTypes.INT())
def add_one(a):
    return a + 1

И flink- python jar должен быть загружен при запуске sql -client, например: 

$ cd $FLINK_HOME/bin
$ ./start-cluster.sh
$ ./sql-client.sh embedded -pyfs xxx.py -j ../opt/flink-python_2.11-1.11.0.jar

Кроме того, вам нужно добавить taskmanager.memory.task.off-heap.size: 79mb в $FLINK_HOME/conf/flink-conf.yaml или другие файлы, которые можно использовать для установки конфигураций (например, файл среды sql клиента ), иначе вы получите ошибку при выполнении python udf: 

[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.TableException: The configured Task Off-Heap Memory 0 bytes is less than the least required Python worker Memory 79 mb. The Task Off-Heap Memory can be configured using the configuration key'taskmanager.memory .task.off-heap.size'.

Best,
Wei

...