雲計算

如何在 Apache Flink 1.10 中使用 Python UDF?

75.jpg
鏡像下載、域名解析、時間同步請點擊 阿里巴巴開源鏡像站

一、安裝 PyFlink

我們需要先安裝 PyFlink,可以通過 PyPI 獲得,並且可以使用 pip install 進行便捷安裝。
注意: 安裝和運行 PyFlink 需要 Python 3.5 或更高版本。

$ python -m pip install apache-Apache Flink

二、定義一個 UDF

除了擴展基類 ScalarFunction 之外,定義 Python UDF 的方法有很多。下面的示例顯示了定義 Python UDF 的不同方法,該函數以 BIGINT 類型的兩列作為輸入參數,並返回它們的和作為結果。

  • Option 1: extending the base class ScalarFunction
class Add(ScalarFunction):
  def eval(self, i, j):
    return i + j
add = udf(Add(), [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT())
  • Option 2: Python function
@udf(input_types=[DataTypes.BIGINT(), DataTypes.BIGINT()], result_type=DataTypes.BIGINT())
def add(i, j):
  return i + j
  • Option 3: lambda function
add = udf(lambda i, j: i + j, [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT())
  • Option 4: callable function
class CallableAdd(object):
  def __call__(self, i, j):
    return i + j
add = udf(CallableAdd(), [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT())
  • Option 5: partial function
return i + j + k
add = udf(functools.partial(partial_add, k=1), [DataTypes.BIGINT(), DataTypes.BIGINT()],
          DataTypes.BIGINT())

三、註冊一個UDF

  • register the Python function
table_env.register_function("add", add)
  • Invoke a Python UDF
my_table.select(```js
"add(a, b)")
  • Example Code

下面是一個使用 Python UDF 的完整示例。

from PyFlink.table import StreamTableEnvironment, DataTypes
from PyFlink.table.descriptors import Schema, OldCsv, FileSystem
from PyFlink.table.udf import udf
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
t_env = StreamTableEnvironment.create(env)
t_env.register_function("add", udf(lambda i, j: i + j, [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT()))
t_env.connect(FileSystem().path('/tmp/input')) \
    .with_format(OldCsv()
                 .field('a', DataTypes.BIGINT())
                 .field('b', DataTypes.BIGINT())) \
    .with_schema(Schema()
                 .field('a', DataTypes.BIGINT())
                 .field('b', DataTypes.BIGINT())) \
    .create_temporary_table('mySource')
t_env.connect(FileSystem().path('/tmp/output')) \
    .with_format(OldCsv()
                 .field('sum', DataTypes.BIGINT())) \
    .with_schema(Schema()
                 .field('sum', DataTypes.BIGINT())) \
    .create_temporary_table('mySink')
t_env.from_path('mySource')\
    .select("add(a, b)") \
    .insert_into('mySink')
t_env.execute("tutorial_job")
  • 提交作業

首先,您需要在“ / tmp / input”文件中準備輸入數據。例如,

$ echo "1,2" > /tmp/input

接下來,您可以在命令行上運行此示例:

$ python python_udf_sum.py

通過該命令可在本地小集群中構建並運行 Python Table API 程序。您還可以使用不同的命令行將 Python Table API 程序提交到遠程集群。
最後,您可以在命令行上查看執行結果:

$ cat /tmp/output
3

四、Python UDF 的依賴管理

在許多情況下,您可能希望在 Python UDF 中導入第三方依賴。下面的示例將指導您如何管理依賴項。
假設您想使用 mpmath 來執行上述示例中兩數的和。Python UDF 邏輯可能如下:

@udf(input_types=[DataTypes.BIGINT(), DataTypes.BIGINT()], result_type=DataTypes.BIGINT())
def add(i, j):
    from mpmath import fadd # add third-party dependency
    return int(fadd(1, 2))

要使其在不包含依賴項的工作節點上運行,可以使用以下 API 指定依賴項:

# echo mpmath==1.1.0 > requirements.txt
# pip download -d cached_dir -r requirements.txt --no-binary :all:
t_env.set_python_requirements("/path/of/requirements.txt", "/path/of/cached_dir")

用戶需要提供一個 requirements.txt 文件,並且在裡面申明使用的第三方依賴。如果無法在群集中安裝依賴項(網絡問題),則可以使用參數“requirements_cached_dir”,指定包含這些依賴項的安裝包的目錄,如上面的示例所示。依賴項將上傳到群集並脫機安裝。
下面是一個使用依賴管理的完整示例:

from PyFlink.datastream import StreamExecutionEnvironment
from PyFlink.table import StreamTableEnvironment, DataTypes
from PyFlink.table.descriptors import Schema, OldCsv, FileSystem
from PyFlink.table.udf import udf
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
t_env = StreamTableEnvironment.create(env)
@udf(input_types=[DataTypes.BIGINT(), DataTypes.BIGINT()], result_type=DataTypes.BIGINT())
def add(i, j):
    from mpmath import fadd
    return int(fadd(1, 2))
t_env.set_python_requirements("/tmp/requirements.txt", "/tmp/cached_dir")
t_env.register_function("add", add)
t_env.connect(FileSystem().path('/tmp/input')) \
    .with_format(OldCsv()
                 .field('a', DataTypes.BIGINT())
                 .field('b', DataTypes.BIGINT())) \
    .with_schema(Schema()
                 .field('a', DataTypes.BIGINT())
                 .field('b', DataTypes.BIGINT())) \
    .create_temporary_table('mySource')
t_env.connect(FileSystem().path('/tmp/output')) \
    .with_format(OldCsv()
                 .field('sum', DataTypes.BIGINT())) \
    .with_schema(Schema()
                 .field('sum', DataTypes.BIGINT())) \
    .create_temporary_table('mySink')
t_env.from_path('mySource')\
    .select("add(a, b)") \
    .insert_into('mySink')
t_env.execute("tutorial_job")
  • 提交作業

首先,您需要在“/ tmp / input”文件中準備輸入數據。例如,

echo "1,2" > /tmp/input
1
2

其次,您可以準備依賴項需求文件和緩存目錄:

$ echo "mpmath==1.1.0" > /tmp/requirements.txt
$ pip download -d /tmp/cached_dir -r /tmp/requirements.txt --no-binary :all:

接下來,您可以在命令行上運行此示例:

$ python python_udf_sum.py

最後,您可以在命令行上查看執行結果:

$ cat /tmp/output
3

五、快速上手

PyFlink 為大家提供了一種非常方便的開發體驗方式 - PyFlink Shell。當成功執行 python -m pip install apache-flink 之後,你可以直接以 pyflink-shell.sh local 來啟動一個 PyFlink Shell 進行開發體驗,如下所示:
1.gif

六、更多場景

不僅僅是簡單的 ETL 場景支持,PyFlink 可以完成很多複雜場的業務場景需求,比如我們最熟悉的雙 11 大屏的場景,如下:
2.jpg

關於上面示例的更多詳細請查閱:https://enjoyment.cool/2019/12/05/Apache-Flink-說道系列-如何在PyFlink-1-10中自定義Python-UDF/

阿里巴巴開源鏡像站 提供全面,高效和穩定的鏡像下載服務。釘釘搜索 ' 21746399 ‘ 加入鏡像站官方用戶交流群。”

Leave a Reply

Your email address will not be published. Required fields are marked *