Pyflink 自定义函数基础实例日志监控

编程入门行业动态更新时间:2024-10-21 06:12:05

01 自定义函数概述

自定义函数是 PyFlink Table API 中最重要的功能之一，其允许用户在 PyFlink Table API 中使用 Python 语言开发的自定义函数，极大地拓宽了 Python Table API 的使用范围。

目前 Python 自定义函数的功能已经非常完善，支持多种类型的自定义函数，比如 UDF（scalar function）、UDTF（table function）、UDAF（aggregate function），UDTAF（table aggregate function)

根据输入 / 输出数据的行数，Flink Table API & SQL 中，自定义函数可以分为以下几类：

自定义函数Single Row InputMultiple Row InputSingle Row OutputScalarFunctionAggregateFunctionMultiple Row OutputTableFunctionTableAggregateFunction

02 标量函数的使用

2.1 扩展 ScalarFunction 基类实现 Python 标量函数

PyFlink是最为简单的单行输入输出的 Python 标量函数。如果要定义 Python 标量函数，可以继承 pyflink.table.udf 中的基类 ScalarFunction，并实现 eval 方法。 Python 标量函数的行为由名为 eval 的方法定义，eval 方法支持可变长参数，例如 eval(* args)。

通过继承 ScalarFunction 的方式来定义 Python UDF 有以下用处：ScalarFunction 的基类 UserDefinedFunction 中定义了一个 open 方法，该方法只在作业初始化时执行一次，因此可以利用该方法，做一些初始化工作，比如加载机器学习模型、连接外部服务等。此外，还可以通过 open 方法中的 function_context 参数，注册及使用 metrics。

以下示例显示了如何定义自己的 Python 哈希函数、如何在 TableEnvironment 中注册它以及如何在作业中使用它。

Python Table API 中使用 Python 自定义函数

from pyflink.table.expressions import call 
from pyflink.table import (EnvironmentSettings, TableEnvironment, DataTypes, TableDescriptor,Schema,FormatDescriptor)
from pyflink.table.udf import udf,ScalarFunction# 继承 pyflink.table.udf 中的基类 ScalarFunction，并实现 eval 方法
class HashCode(ScalarFunction):def __init__(self):self.factor = 12def eval(self, s):return hash(s) * self.factordef hashCode_with_udf_from_elem():t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())# 定义输入 sourcetab = t_env.from_elements(elements=[('first',1),('second',2),('third',3),('forth',4)],schema=['name','val'])# 定义输出 sink ，`connector` 为 print 直接终端打印输出t_env.create_temporary_table('sink',TableDescriptor.for_connector('print').schema(Schema.new_builder().column('name', DataTypes.STRING()).column('val', DataTypes.BIGINT()).column('hashcode', DataTypes.BIGINT()).build()).build())# 实例化子类 定义 udfhash_code = udf(HashCode(), result_type=DataTypes.BIGINT())# 在 Python Table API 中使用 Python 自定义函数，有如下两种调用方式# 1. hash_code(tab.val)tab = tab.select(tab.name, tab.val, hash_code(tab.val))# 2. call(hash_code,tab.val)# tab = tab.select(tab.name, tab.val, call(hash_code,tab.val))tab.execute_insert('sink').wait()if __name__ == '__main__':hashCode_with_udf_from_elem()

SQL API 中使用 Python 自定义函数

Table API 实现中我们采用了通过列表类型的对象创建输入表，通过 Catalog 创建输出表

这里，我们通过 DDL 创建输入表和输出表，并展示在 SQL API 中使用 Python 自定义函数

from pyflink.table.expressions import call 
from pyflink.table import (EnvironmentSettings, TableEnvironment, DataTypes, TableDescriptor,Schema, FormatDescriptor)
from pyflink.table.udf import udf,ScalarFunction# 继承 pyflink.table.udf 中的基类 ScalarFunction，并实现 eval 方法
class HashCode(ScalarFunction):def __init__(self):self.factor = 12def eval(self, s):return hash(s) * self.factordef hashCode_with_udf_from_elem():t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())# 定义输入 sourcesource = t_env.from_elements(elements=[('first',1),('second',2),('third',3),('forth',4)],schema=['name','val'])# 定义输出 sink ，`connector` 为 print 直接终端打印输出sink_ddl = """create table sink (name STRING,val BIGINT,`hashCode` BIGINT) with ('connector' = 'print')"""t_env.execute_sql(sink_ddl)# define the udf# 在 SQL API 中使用 Python 自定义函数t_env.create_temporary_function("hash_code", udf(HashCode(), result_type=DataTypes.BIGINT()))t_env.execute_sql("INSERT INTO sink SELECT name, val, hash_code(val) FROM %s" %source).wait()if __name__ == '__main__':hashCode_with_udf_from_elem()

2.2 其他方式定义 Python 标量函数

除了扩展基类 ScalarFunction 之外，还支持多种方式来定义 Python 标量函数：

方式一：扩展基类 calarFunction方式二：普通 Python 函数方式三：lambda 函数方式四：callable 函数方式五：partial 函数

不同的定义方式都有一个共同的特点就是：需要通过名字为 udf 的装饰器，声明这是一个 scalar function；需要通过装饰器中的 result_type 参数，声明 scalar function 的结果类型。

以下示例显示了上述多种定义 Python 标量函数的方式。该函数需要两个类型为 bigint 的参数作为输入参数，并返回它们的总和作为结果。

# 方式一：扩展基类 calarFunction
class Add(ScalarFunction):def eval(self, i, j):return i + jadd = udf(Add(), result_type=DataTypes.BIGINT())# 方式二：普通 Python 函数
@udf(result_type=DataTypes.BIGINT())
def add(i, j):return i + j# 方式三：lambda 函数
add = udf(lambda i, j: i + j, result_type=DataTypes.BIGINT())# 方式四：callable 函数
class CallableAdd(object):def __call__(self, i, j):return i + jadd = udf(CallableAdd(), result_type=DataTypes.BIGINT())# 方式五：partial 函数
def partial_add(i, j, k):return i + j + kadd = udf(functools.partial(partial_add, k=1), result_type=DataTypes.BIGINT())# 注册 Python 自定义函数
table_env.create_temporary_function("add", add)
# 在 Python Table API 中使用 Python 自定义函数
my_table.select("add(a, b)")# 也可以在 Python Table API 中直接使用 Python 自定义函数
my_table.select(add(my_table.a, my_table.b))

一个上述多种定义 Python 标量函数方式的实例如下：

import functools
from pyflink.table.expressions import call 
from pyflink.table import (EnvironmentSettings, TableEnvironment, DataTypes, TableDescriptor,Schema, FormatDescriptor)
from pyflink.table.udf import udf,ScalarFunction# 方式一：扩展基类 calarFunction
class Add(ScalarFunction):def eval(self, i, j):return i + j# 方式二：普通 Python 函数
@udf(result_type=DataTypes.BIGINT())
def sub(i, j):return i - j# 方式三：lambda 函数
mul = udf(lambda i, j: i * j, result_type=DataTypes.BIGINT())# 方式四：callable 函数
class CallableDiv(object):def __call__(self, i, j):return i / j# 方式五：partial 函数
def partial_mod(i, j):return i % jdef udf_define():t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())# define the sourcesource = t_env.from_elements(elements=[(1,1),(4,2),(6,3)],schema=['a','b'])# define the multiple sink tables# 存储加法UDF计算结果t_env.create_temporary_table('add_sink',TableDescriptor.for_connector('print').schema(Schema.new_builder().column('a', DataTypes.BIGINT()).column('b', DataTypes.BIGINT()).column('res', DataTypes.BIGINT()).build()).build())# 存储减法UDF计算结果t_env.create_temporary_table('sub_sink',TableDescriptor.for_connector('print').schema(Schema.new_builder().column('a', DataTypes.BIGINT()).column('b', DataTypes.BIGINT()).column('res', DataTypes.BIGINT()).build()).build())# 存储乘法UDF计算结果t_env.create_temporary_table('mul_sink',TableDescriptor.for_connector('print').schema(Schema.new_builder().column('a', DataTypes.BIGINT()).column('b', DataTypes.BIGINT()).column('res', DataTypes.BIGINT()).build()).build())# 存储除法UDF计算结果t_env.create_temporary_table('div_sink',TableDescriptor.for_connector('print').schema(Schema.new_builder().column('a', DataTypes.BIGINT()).column('b', DataTypes.BIGINT()).column('res', DataTypes.BIGINT()).build()).build())# 存储取余UDF计算结果t_env.create_temporary_table('mod_sink',TableDescriptor.for_connector('print').schema(Schema.new_builder().column('a', DataTypes.BIGINT()).column('b', DataTypes.BIGINT()).column('res', DataTypes.BIGINT()).build()).build())# 创建 statement setstatement_set = t_env.create_statement_set()# 将加法UDF计算结果写入 "add_sink"add = udf(Add(), result_type=DataTypes.BIGINT())tab = source.select(source.a, source.b, add(source.a,source.b))statement_set.add_insert("add_sink", tab)
89 syslog-tcp:0:89: key=None value=b'polkitd[834]: Unregistered Authentication Agent for unix-process:65042:103999247 (system bus name :1.809, object path //freedesktop/PolicyKit1/AuthenticationAgent, locale zh_CN.UTF-8) (disconnected from bus)'# 将减法UDF计算结果写入 "sub_sink"tab = source.select(source.a, source.b, sub(source.a,source.b))statement_set.add_insert("sub_sink", tab)# 将乘法UDF计算结果写入 "mul_sink"tab = source.select(source.a, source.b, mul(source.a,source.b))statement_set.add_insert("mul_sink", tab)# 将除法UDF计算结果写入 "div_sink"div = udf(CallableDiv(), result_type=DataTypes.BIGINT())tab = source.select(source.a, source.b, div(source.a,source.b))statement_set.add_insert("div_sink", tab)# 将取余UDF计算结果写入 "mod_sink"mod = udf(functools.partial(partial_mod), result_type=DataTypes.BIGINT())tab = source.select(source.a, source.b, mod(source.a,source.b))statement_set.add_insert("mod_sink", tab)# 执行 statement setstatement_set.execute().wait()if __name__ == '__main__':udf_define()

03 Python 依赖管理

在定义 UDF 时，我们可能需要使用到一些第三方库。如果将编写的任务提交到远端集群中运行时，就需要在运行环境中下载相同的第三方库。

为此 PyFlink 提供了指定第三方库的方式

3.1 根据第三方库安装包指定依赖

Python Table API 提供了如下方式指定第三方库

table_env.add_python_file(file_path)

Python DataStream API 提供了相似的方式指定第三方库

stream_execution_environment.add_python_file(file_path)

3.2 根据依赖描述文件指定依赖

PyFlink 还允许指定一个描述第三方 Python 依赖项的 requirements.txt 文件来导入第三方库

这些 Python 依赖项将安装到工作目录中并添加到 Python UDF 工作器的 PYTHONPATH 中

Python Table API 提供了两种方式来导入：

# 方式1：直接传入 requirements.txt ，当集群可以联网时，会下载这些依赖项
table_env.set_python_requirements(requirements_file_path="/path/to/requirements.txt")# 方式2：当集群不能联网时，可以先准备好一个由 requirements.txt 生成的包含有安装包的依赖文件夹 cached_dir ，集群会离线安装这些依赖项
table_env.set_python_requirements(requirements_file_path="/path/to/requirements.txt",requirements_cache_dir="cached_dir")

方式 2 相比于方式 1 有两点好处：将依赖下载与作业执行解耦，提高了作业执行的效率；依赖项的安装包预先下载好，适用于集群不能联网的情况。

Python DataStream API 也可以通过相同的方式导入第三方库：

stream_execution_environment.set_python_requirements(requirements_file_path="/path/to/requirements.txt",requirements_cache_dir="cached_dir")

对于集群中无法访问的依赖，可以通过 requirements_cached_dir 参数指定一个包含这些依赖的安装包的目录。通过如下命令生成包含有安装包的 cached_dir 文件夹，并将其上传到集群支持离线安装。

pip download -d cached_dir -r requirements.txt --no-binary :all:

04 日志监控告警实例

本实例通过 Flink 结合 UDF 对系统上报的日志进行实时解析并生成告警，搭建实时日志监控系统。实例中展示了如何在 Flink 中使用自定义函数 UDF 、第三方的依赖包，来实现复杂的日志解析逻辑。

4.1 根据数据对象创建输入输出表

本案例的数据对象就是普通的数据日志，包含日志类型 topic、ip源 ip_src、主机IP ip_host、用户名称 user_name、用户组 user_group、日志内容 msg_content、日志时间戳 msg_time，如下所示：

88 syslog-tcp:0:88: key=None value=b'systemd: Started System Logging Service.'
90 syslog-tcp:0:90: key=None value=b'root: "root pts/4 2020-10-26 15:06 (172.25.2.21) ======================================= is login "'
91 syslog-tcp:0:91: key=None value=b'root: [ root pts/4 2020-10-26 15:06 (172.25.2.21)]# "2020-10-26 15:11:44 root pts/4 172.25.2.21 source /etc/profile"'

读取原始日志数据，输入格式使用 raw，根据数据对象创建如下输入表：

t_env.create_temporary_table('source',TableDescriptor.for_connector('filesystem').schema(Schema.new_builder().column('line', DataTypes.STRING()).build()).option('path', input_path).format('raw').build())

我们将解析后的日志按照字段存入输出表：

t_env.create_temporary_table('sink',TableDescriptor.for_connector('filesystem').schema(Schema.new_builder().column('topic', DataTypes.STRING()).column('fake_country', DataTypes.STRING()).column('ip_src', DataTypes.STRING()).column('ip_host', DataTypes.STRING()).column('user_name', DataTypes.STRING()).column('user_group', DataTypes.STRING()).column('msg_content', DataTypes.STRING()).column('msg_time', DataTypes.STRING()).build()).option('path', output_path).format(FormatDescriptor.for_format('canal-json').build()).build())

4.2 定义 UDF 解析日志

定义 UDF 解析日志，提取日志中的各字段：日志类型 topic、ip源 ip_src、主机IP ip_host、用户名称 user_name、用户组 user_group、日志内容 msg_content、日志时间戳 msg_time

1 生成随机国家

使用第三方库 faker 生成随机中文国家名称

import faker
@udf(input_types=[], result_type=DataTypes.STRING())
def get_fake_country():return faker.Faker(locale='zh_CN').country()

2 获取日志类型

我们将所有的日志记录划分为 3 种日志类型： syslog-iptables, syslog-user, syslog-system

@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.STRING())
def get_topic(line):if 'IN=' in line and 'OUT=' in line and 'MAC=' in line:return 'syslog-iptables'elif '=======================================' in line or re.search(r'localhost (.+?): \[', line, re.M | re.I):return 'syslog-user'else:return 'syslog-system'

3 获取主机IP地址

使用正则表达式获取日志记录中的主机IP地址，ip:{pattern_ip}

@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.STRING())
def get_ip_host(line):pattern_ip = r'(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}' \r'|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])'pattern_ip_host = f'ip:{pattern_ip}'ip_host = re.search(pattern_ip_host, line, re.M | re.I) or ''if ip_host:ip_host = ip_host.group()[3:]return ip_host

4 获取IP源

当用户登录时，日志中会写入其访问IP，我们使用正则表达式获取日志记录中的IP源

@udf(input_types=[DataTypes.STRING(), DataTypes.STRING()], result_type=DataTypes.STRING())
def get_ip_src(topic, line):pattern_ip = r'(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}' \r'|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])'pattern_time1 = '[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'  # YYYY-mm-dd HH:MM:SSpattern_ip_src = rf'{pattern_time1} (?:({pattern_ip})|(\({pattern_ip}\)))'if topic == 'syslog-user':ip_src = re.search(pattern_ip_src, line, re.M | re.I) or ''if ip_src:ip_src = ip_src.group().split(' ')[2]else:ip_src = ''return ip_src

5 获取用户名称

用户名称也是在用户登录时，日志中会写入其用户名称，我们使用正则表达式获取日志记录中的用户名称

@udf(input_types=[DataTypes.STRING(), DataTypes.STRING()], result_type=DataTypes.STRING())
def get_user_name(topic, line):pattern_ip = r'(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}' \r'|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])'pattern_time1 = '[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'  # YYYY-mm-dd HH:MM:SSpattern_char = '(.+?)'  # 字符pattern_user = f'{pattern_time1} {pattern_ip} {pattern_char} '  # 注意后面有个空格if topic == 'syslog-user':user_name = re.search(pattern_user, line, re.M | re.I) or ''if user_name:user_name = user_name.group().split(' ')[-2]else:user_name = ''return user_name

6 获取用户组

用户组信息也包含在用户登录日志中

@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.STRING())
def get_user_group(line):pattern_char = '(.+?)'  # 字符pattern_group = f'localhost {pattern_char}:'user_group = re.search(pattern_group, line, re.M | re.I) or ''if user_group:user_group = user_group.group().split(' ')[-1][:-1]return user_group

7 获取日志内容

根据日志类型使用正则表达式去提取不同的日志内容

@udf(input_types=[DataTypes.STRING(), DataTypes.STRING()], result_type=DataTypes.STRING())
def get_msg_content(topic, line):import repattern_ip = r'(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}' \r'|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1}|[1-9]{1}\d{1}|1\d\d|2[0-4]\d|25[0-5])'pattern_char = '(.+?)'  # 字符pattern_time1 = '[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'  # YYYY-mm-dd HH:MM:SSpattern_user = f'{pattern_time1} {pattern_ip} {pattern_char} '  # 注意后面有个空格pattern_content_login = f'======================================= is login'  # 用户登录信息pattern_content_user = f'{pattern_user}{pattern_char}"\'$'  # 用户命令信息，注意以 "' 2个字符结尾  pattern_content_system = f'localhost {pattern_char}\'$'  # 系统命令信息，注意以 ' 1个字符结尾  pattern_content_iptables = f'localhost(.*?)IN=(.*?)OUT=(.*?)MAC=(.*?)\'$'  # 防火墙命令信息，注意以 ' 1个字符结尾  pattern_group = f'localhost {pattern_char}:'user_group = re.search(pattern_group, line, re.M | re.I) or ''if user_group:user_group = user_group.group().split(' ')[-1][:-1]if topic == 'syslog-user':msg_content = re.search(pattern_content_user, line, re.M | re.I) or ''if msg_content:msg_content = ' '.join(msg_content.group().split(' ')[4:])[:-2]  # 注意以 "' 2个字符结尾  if not msg_content and pattern_content_login in line:msg_content = 'login'elif topic == 'syslog-system':msg_content = re.search(pattern_content_system, line, re.M | re.I) or ''if msg_content:msg_content = msg_content.group()[10:-1]  # 去掉 localhost 和空格，注意以 ' 1个字符结尾  if msg_content.startswith(user_group):msg_content = msg_content[(len(user_group) + 2):]  # 去掉用户组，以及冒号和空格elif topic == 'syslog-iptables':msg_content = re.search(pattern_content_iptables, line, re.M | re.I) or ''if msg_content:msg_content = msg_content.group()[10:-1]  # 去掉 localhost 和空格，注意以 ' 1个字符结尾if msg_content.startswith(user_group):msg_content = msg_content[(len(user_group) + 2):]  # 去掉用户组，以及冒号和空格else:msg_content = ''return msg_content

8 获取日志时间戳

@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.STRING())
def get_msg_time(line):pattern_time2 = '(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sept|Oct|Nov|Dec) ' \'[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'  # Jun dd HH:MM:SS dict_month = {'Jan': '1','Feb': '2','Mar': '3','Apr': '4','May': '5','Jun': '6','Jul': '7','Aug': '8','Sept': '9','Oct': '10','Nov': '11','Dec': '12'}msg_time = re.search(pattern_time2, line, re.M | re.I) or ''if msg_time:msg_time = msg_time.group()year = str(datetime.now().year)  # 没有年份的话用今年来代替month = dict_month[msg_time.split(" ")[0]]other = ' '.join(msg_time.split(" ")[1:])msg_time = f'{year}-{month}-{other}'  # 格式化为 YYYY-mm-ddreturn msg_time