PyFlink 教程（三）：PyFlink DataStream

2021-06-09  本文已影响0人  Flink中文社区

一、背景

Flink 1.13 已于近期正式发布，超过 200 名贡献者参与了 Flink 1.13 的开发，提交了超过 1000 个 commits，完成了若干重要功能。其中，PyFlink 模块在该版本中也新增了若干重要功能，比如支持了 state、自定义 window、row-based operation 等。随着这些功能的引入，PyFlink 功能已经日趋完善，用户可以使用 Python 语言完成绝大多数类型Flink作业的开发。接下来，我们详细介绍如何在 Python DataStream API 中使用 state & timer 功能。

二、state 功能介绍

作为流计算引擎，state 是 Flink 中最核心的功能之一。

state 使用示例

如下是一个简单的示例，说明如何在 Python DataStream API 作业中使用 state：

from pyflink.common import WatermarkStrategy, Row
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import NumberSequenceSource
from pyflink.datastream.functions import RuntimeContext, MapFunction
from pyflink.datastream.state import ValueStateDescriptor


class MyMapFunction(MapFunction):

    def open(self, runtime_context: RuntimeContext):
        state_desc = ValueStateDescriptor('cnt', Types.LONG())
        # 定义value state
        self.cnt_state = runtime_context.get_state(state_desc)

    def map(self, value):
        cnt = self.cnt_state.value()
        if cnt is None:
            cnt = 0

        new_cnt = cnt + 1
        self.cnt_state.update(new_cnt)
        return value[0], new_cnt


def state_access_demo():
    # 1. 创建 StreamExecutionEnvironment
    env = StreamExecutionEnvironment.get_execution_environment()

    # 2. 创建数据源
    seq_num_source = NumberSequenceSource(1, 100)
    ds = env.from_source(
        source=seq_num_source,
        watermark_strategy=WatermarkStrategy.for_monotonous_timestamps(),
        source_name='seq_num_source',
        type_info=Types.LONG())

    # 3. 定义执行逻辑
    ds = ds.map(lambda a: Row(a % 4, 1), output_type=Types.ROW([Types.LONG(), Types.LONG()])) \
           .key_by(lambda a: a[0]) \
           .map(MyMapFunction(), output_type=Types.TUPLE([Types.LONG(), Types.LONG()]))

    # 4. 将打印结果数据
    ds.print()

    # 5. 执行作业
    env.execute()


if __name__ == '__main__':
    state_access_demo()

在上面的例子中，我们定义了一个 MapFunction，该 MapFunction 中定义了一个名字为 “cnt_state” 的 ValueState，用于记录每一个 key 出现的次数。

说明：

ds1 = ...  # type DataStream
ds2 = ...  # type DataStream
ds1.connect(ds2) \
    .key_by(key_selector1=lambda a: a[0], key_selector2=lambda a: a[0]) \
    .map(MyCoMapFunction())  # 可以在MyCoMapFunction中使用state

可以使用 state 的 API 列表如下：

操作 自定义函数
KeyedStream map MapFunction
flat_map FlatMapFunction
reduce ReduceFunction
filter FilterFunction
process KeyedProcessFunction
ConnectedStreams map CoMapFunction
flat_map CoFlatMapFunction
process KeyedCoProcessFunction
WindowedStream apply WindowFunction
process ProcessWindowFunction

state 工作原理

上图是 PyFlink 中，state 工作原理的架构图。从图中我们可以看出，Python 自定义函数运行在 Python worker 进程中，而 state backend 运行在 JVM 进程中（由 Java 算子来管理）。当 Python 自定义函数需要访问 state 时，会通过远程调用的方式，访问 state backend。

我们知道，远程调用的开销是非常大的，为了提升 state 读写的性能，PyFlink 针对 state 读写做了以下几个方面的优化工作：

其中 LRU cache 可以细分为二级，如下图所示：

image.png

说明：

工作流程：

state 性能调优

通过前一节的介绍，我们知道 PyFlink 使用了多种优化手段，用于提升 state 读写的性能，这些优化行为可以通过以下参数配置：

配置 说明
python.state.cache-size Python worker 中读 cache 以及写 cache 的大小。（二级 cache）需要注意的是：读 cache、写 cache是独立的，当前不支持分别配置读 cache 以及写 cache 的大小。
python.map-state.iterate-response-batch-size 当遍历 MapState 时，每次从 state backend 读取并返回给 Python worker 的 entry 的最大个数。
python.map-state.read-cache-size 一个 MapState 的读 cache 中最大允许的 entry 个数（一级 cache）。当一个 MapState 中，读 cache 中的 entry 个数超过该阈值时，会通过 LRU 策略从读 cache 中删除最近最少访问过的 entry。
python.map-state.write-cache-size 一个 MapState 的写 cache 中最大允许的待更新 entry 的个数（一级 cache）。当一个 MapState 中，写 cache 中待更新的 entry 的个数超过该阈值时，会将该 MapState 下所有待更新 state 数据写回远端的 state backend。

需要注意的是，state 读写的性能不仅取决于以上参数，还受其它因素的影响，比如：

三、timer 功能介绍

timer 使用示例

除了 state 之外，用户还可以在 Python DataStream API 中使用定时器 timer。

import datetime

from pyflink.common import Row, WatermarkStrategy
from pyflink.common.typeinfo import Types
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.functions import KeyedProcessFunction, RuntimeContext
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.table import StreamTableEnvironment


class CountWithTimeoutFunction(KeyedProcessFunction):

    def __init__(self):
        self.state = None

    def open(self, runtime_context: RuntimeContext):
        self.state = runtime_context.get_state(ValueStateDescriptor(
            "my_state", Types.ROW([Types.STRING(), Types.LONG(), Types.LONG()])))

    def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
        # retrieve the current count
        current = self.state.value()
        if current is None:
            current = Row(value.f1, 0, 0)

        # update the state's count
        current[1] += 1

        # set the state's timestamp to the record's assigned event time timestamp
        current[2] = ctx.timestamp()

        # write the state back
        self.state.update(current)

        # schedule the next timer 60 seconds from the current event time
        ctx.timer_service().register_event_time_timer(current[2] + 60000)

    def on_timer(self, timestamp: int, ctx: 'KeyedProcessFunction.OnTimerContext'):
        # get the state for the key that scheduled the timer
        result = self.state.value()

        # check if this is an outdated timer or the latest timer
        if timestamp == result[2] + 60000:
            # emit the state on timeout
            yield result[0], result[1]


class MyTimestampAssigner(TimestampAssigner):

    def __init__(self):
        self.epoch = datetime.datetime.utcfromtimestamp(0)

    def extract_timestamp(self, value, record_timestamp) -> int:
        return int((value[0] - self.epoch).total_seconds() * 1000)


if __name__ == '__main__':
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(stream_execution_environment=env)

    t_env.execute_sql("""
            CREATE TABLE my_source (
              a TIMESTAMP(3),
              b VARCHAR,
              c VARCHAR
            ) WITH (
              'connector' = 'datagen',
              'rows-per-second' = '10'
            )
        """)

    stream = t_env.to_append_stream(
        t_env.from_path('my_source'),
        Types.ROW([Types.SQL_TIMESTAMP(), Types.STRING(), Types.STRING()]))
    watermarked_stream = stream.assign_timestamps_and_watermarks(
        WatermarkStrategy.for_monotonous_timestamps()
                         .with_timestamp_assigner(MyTimestampAssigner()))

    # apply the process function onto a keyed stream
    watermarked_stream.key_by(lambda value: value[1])\
        .process(CountWithTimeoutFunction()) \
        .print()

    env.execute()

在上述示例中，我们定义了一个 KeyedProcessFunction，该 KeyedProcessFunction 记录每一个 key 出现的次数，当一个 key 超过 60 秒没有更新时，会将该 key 以及其出现次数，发送到下游节点。

除了 event time timer 之外，用户还可以使用 processing time timer。

timer 工作原理

timer 的工作流程是这样的：

需要注意的是：由于 timer 注册消息以及触发消息通过数据通道异步地在 Java 算子以及 Python worker 之间传输，这会造成在某些场景下，timer 的触发可能没有那么及时。比如当用户注册了一个 processing time timer，当 timer 触发之后，触发消息通过数据通道传输到 Python UDF 时，可能已经是几秒中之后了。

四、总结

在这篇文章中，我们主要介绍了如何在 Python DataStream API 作业中使用 state & timer，state & timer 的工作原理以及如何进行性能调优。接下来，我们会继续推出 PyFlink 系列文章，帮助 PyFlink 用户深入了解 PyFlink 中各种功能、应用场景以及最佳实践等。

上一篇下一篇

猜你喜欢

热点阅读