Operators

DataStream Transformations #

DataStream 程序 在Flink里是一个常规操作,对数据流进行转换(mapping, filtering, reducing)

Functions

转换接收用户定义的函数作为输入, 以定义转换的功能

Implementing Function Interfaces #

不同的函数接口用于不同的转换在 Python DataStream API

比如MapFunction 是用于map transformation, 

FilterFunction 是用于filter transformation

[root@master pyflink]# cat t104.py 
from pyflink.common import WatermarkStrategy, Row
from pyflink.common.serialization import Encoder
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FileSink, OutputFileConfig, NumberSequenceSource
from pyflink.datastream.functions import RuntimeContext, MapFunction
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(stream_execution_environment=env)
class MyMapFunction(MapFunction):
    
    def map(self, value):
        return value + 1
        
data_stream = env.from_collection([1, 2, 3, 4, 5], type_info=Types.INT())
mapped_stream = data_stream.map(MyMapFunction(), output_type=Types.INT())
data_stream.print()
mapped_stream.print()
env.execute()

Lambda Function #

如下所示,转换还可以接收lambda函数来定义转换的功能：

[root@master pyflink]# cat t105.py 
from pyflink.common import WatermarkStrategy, Row
from pyflink.common.serialization import Encoder
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FileSink, OutputFileConfig, NumberSequenceSource
from pyflink.datastream.functions import RuntimeContext, MapFunction
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(stream_execution_environment=env)
data_stream = env.from_collection([1, 2, 3, 4, 5], type_info=Types.INT())
mapped_stream = data_stream.map(lambda x: x + 1, output_type=Types.INT())
data_stream.print()
mapped_stream.print()
env.execute()
[root@master pyflink]# python t105.py 
2> 1
2> 5
3> 2
4> 3
1> 4
2> 2
2> 6
1> 5
3> 3
4> 4

Output Type #

用户可以在Python DataStream API中显示指定转换的输出类型信息。

Convert DataStream into Table #