Python编程指南（流）Beta - 《Flink 中文文档 - 帮助手册 - 教程》

Jython框架
- 约束
流程序示例
程序框架
项目设置
懒惰的评价
转换
将函数传递给Flink
数据类型
- 元组/列表
数据源
数据接收
并行执行
- 运行环境级别
- 系统级别
执行计划

译者：flink.sojb.cn

Flink中的分析流程序是实现流数据集转换（例如，Filter，映射，Join，分组）的常规程序。流数据集最初是从某些源创建的（例如，通过读取文件或从集合中创建）。结果通过接收器返回，接收器可以例如将数据写入（分布式）文件或标准输出（例如命令行终端）。Flink流处理程序可在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在许多计算机的集群上执行。

为了创建自己的Flink流处理程序，我们建议您从程序框架开始，逐步添加自己的转换。其余部分充当其他算子操作和高级函数的参考。

Jython框架

Flink Python流API使用Jython框架（请参阅http://www.jython.org/archive/21/docs/whatis.html）来驱动给定脚本的执行。Python流层实际上是现有Java流API的薄打包层。

约束

使用Jython有两个主要限制：

最新的Python支持版本是2.7
使用Python C扩展并不简单

流程序示例

以下流处理程序是WordCount的完整工作示例。您可以复制并粘贴代码以在本地运行它（请参阅本节后面的注释）。它计算句子流中每个单词的数量（不区分大小写），窗口大小为50毫秒，并将结果打印到标准输出中。

from org.apache.flink.streaming.api.functions.source import SourceFunction
from org.apache.flink.api.common.functions import FlatMapFunction, ReduceFunction
from org.apache.flink.api.java.functions import KeySelector
from org.apache.flink.streaming.api.windowing.time.Time import milliseconds
class Generator(SourceFunction):
    def __init__(self, num_iters):
        self._running = True
        self._num_iters = num_iters
    def run(self, ctx):
        counter = 0
        while self._running and counter < self._num_iters:
            ctx.collect('Hello World')
            counter += 1
    def cancel(self):
        self._running = False
class Tokenizer(FlatMapFunction):
    def flatMap(self, value, collector):
        for word in value.lower().split():
            collector.collect((1, word))
class Selector(KeySelector):
    def getKey(self, input):
        return input[1]
class Sum(ReduceFunction):
    def reduce(self, input1, input2):
        count1, word1 = input1
        count2, word2 = input2
        return (count1 + count2, word1)
def main(factory):
    env = factory.get_execution_environment()
    env.create_python_source(Generator(num_iters=1000)) \
        .flat_map(Tokenizer()) \
        .key_by(Selector()) \
        .time_window(milliseconds(50)) \
        .reduce(Sum()) \
        .output()
    env.execute()

笔记：

在多节点群集上执行需要共享介质存储，需要预先配置（.eg HDFS）。
给定脚本的输出定向到标准输出。因此，输出将写入相应的工作.out文件。如果脚本在IntelliJ IDE中执行，则输出将显示在控制台选项卡中。

程序框架

正如我们在示例中看到的那样，Flink流程序看起来像普通的Python程序。每个程序包含相同的基本部分：

一个main(factory)函数定义，带有一个环境工厂参数 - 程序入口点，
Environment从工厂获得，
加载/创建初始数据，
指定此数据的转换，
指定计算结果的放置位置，和
执行你的程序。

我们现在将概述每个步骤，但请参阅相应部分以获取更多详细信息。

该main(factory)函数是必须的，Flink执行层使用它来运行给定的Python流程序。

这Environment是所有Flink计划的基础。您可以使用工厂提供的工厂方法获得一个：

factory.get_execution_environment()

为了指定数据源，流运行环境有几种方法。要将文本文件作为一系列行读取，您可以使用：

env = factory.get_execution_environment()
text = env.read_text_file("file:///path/to/file")

这将为您提供一个DataStream，然后您可以在其上应用转换。有关数据源和输入格式的更多信息，请参阅数据源。

拥有DataStream后，您可以应用转换来创建新的DataStream，然后可以将其写入文件，再次转换或与其他DataStream结合使用。您可以通过使用自己的自定义转换函数调用DataStream上的方法来应用转换。例如，Map转换如下所示：

class Doubler(MapFunction):
    def map(self, value):
        return value * 2
data.map(Doubler())

这将通过将原始DataStream中的每个值加倍来创建新的DataStream。有关更多信息和所有转换的列表，请参阅转换。

一旦有了需要写入磁盘的DataStream，就可以在DataStream上调用其中一个方法：

data.write_as_text("<file-path>")
data.write_as_text("<file-path>", mode=WriteMode.OVERWRITE)
data.output()

最后一种方法仅对本地机器上的开发/调试有用，它会将DataSet的内容输出到标准输出。（请注意，在集群中，结果将转到集群节点的标准输出流，最终会出现在工作程序的.out文件中）。前两个顾名思义。有关写入文件的更多信息，请参阅数据接收器。

一旦您指定的完整程序，你需要调用execute的Environment。这将在本地计算机上执行或提交程序以在群集上执行，具体取决于Flink的启动方式。

项目设置

除了设置Flink外，无需额外的工作。使用Jython执行Python脚本意味着不需要外部包，程序就像是一个jar文件一样执行。

Python API在Windows / Linux / OSX系统上进行了测试。

懒惰的评价

所有Flink程序都是懒惰地执行：当执行程序的main方法时，数据加载和转换不会直接发生。而是创建每个算子操作并将其添加到程序的计划中。当execute()在Environment对象上调用其中一个方法时，实际执行这些算子操作。程序是在本地执行还是在集群上执行取决于程序的环境。

懒惰的评估使您可以构建Flink作为一个整体计划单元执行的复杂程序。

转换

数据转换将一个或多个DataStream转换为新的DataStream。程序可以将多个转换组合到复杂的程序集中。

本节简要概述了可用的转换。该转换文档与示例全部转换的完整描述。

转换：Map PythonDataStream→PythonDataStream

描述：采用一个数据元并生成一个数据元。

class Doubler(MapFunction):
    def map(self, value):
        return value * 2
data_stream.map(Doubler())

转换：FlatMap PythonDataStream→PythonDataStream

描述：采用一个数据元并生成零个，一个或多个数据元。

class Tokenizer(FlatMapFunction):
    def flatMap(self, word, collector):
        collector.collect((1, word))
data_stream.flat_map(Tokenizer())

转换：Filter PythonDataStream→PythonDataStream

描述：计算每个数据元的布尔函数，并保存函数返回true的数据元。

class GreaterThen1000(FilterFunction):
    def filter(self, value):
        return value &gt; 1000
data_stream.filter(GreaterThen1000())

转换：KeyBy PythonDataStream→PythonKeyedStream

描述：逻辑上将流分区为不相交的分区，每个分区包含相同Keys的数据元。在内部，这是通过散列分区实现的。见键如何指定键。此转换返回PythonKeyedDataStream。

class Selector(KeySelector):
    def getKey(self, input):
        return input[1]  # Key by the second element in a tuple
data_stream.key_by(Selector()) // Key by field "someKey"

转换：Reduce PythonKeyedStream→PythonDataStream

描述：被Keys化数据流上的“滚动”Reduce。将当前数据元与最后一个Reduce的值组合并发出新值。

class Sum(ReduceFunction):
    def reduce(self, input1, input2):
        count1, val1 = input1
        count2, val2 = input2
        return (count1 + count2, val1)
data.reduce(Sum())

转换：Window PythonKeyedStream→PythonWindowedStream

描述：可以在已经分区的KeyedStream上定义Windows。Windows根据某些特征（例如，在最后5秒内到达的数据）对每个Keys中的数据进行分组。有关窗口的完整说明，请参见windows。

keyed_stream.count_window(10, 5)  # Last 10 elements, sliding (jumping) by 5 elements
keyed_stream.time_window(milliseconds(30))  # Last 30 milliseconds of data
keted_stream.time_window(milliseconds(100), milliseconds(20))  # Last 100 milliseconds of data, sliding (jumping) by 20 milliseconds

转换：Window Apply PythonWindowedStream→PythonDataStream

描述：将一般函数应用于整个窗口。下面是一个手动求和窗口数据元的函数。

class WindowSum(WindowFunction):
    def apply(self, key, window, values, collector):
        sum = 0
        for value in values:
            sum += value[0]
        collector.collect((key, sum))
windowed_stream.apply(WindowSum())

转换：Window Reduce PythonWindowedStream→PythonDataStream

描述：将函数缩减函数应用于窗口并返回缩小的值。

class Sum(ReduceFunction):
    def reduce(self, input1, input2):
        count1, val1 = input1
        count2, val2 = input2
        return (count1 + count2, val1)
windowed_stream.reduce(Sum())

转换：Union PythonDataStream *→PythonDataStream

描述：两个或多个数据流的联合，创建包含来自所有流的所有数据元的新流。注意：如果将数据流与自身联合，则会在结果流中获取两次数据元。

data_stream.union(other_stream1, other_stream2, ...);

转换：Split PythonDataStream→PythonSplitStream

描述：根据某些标准将流拆分为两个或更多个流。

class StreamSelector(OutputSelector):
    def select(self, value):
        return ["even"] if value % 2 == 0 else ["odd"]
splited_stream = data_stream.split(StreamSelector())

转换：Select SplitStream→DataStream

描述：从拆分流中选择一个或多个流。

even_data_stream = splited_stream.select("even")
odd_data_stream = splited_stream.select("odd")
all_data_stream = splited_stream.select("even", "odd")

转换：Iterate PythonDataStream→PythonIterativeStream→PythonDataStream

描述：通过将一个算子的输出重定向到某个先前的算子，在流中创建“反馈”循环。这对于定义不断更新模型的算法特别有用。以下代码以流开头并连续应用迭代体。大于0的数据元将被发送回反馈通道，其余数据元将向下游转发。有关完整说明，请参阅迭代。

class MinusOne(MapFunction):
    def map(self, value):
        return value - 1
class PositiveNumber(FilterFunction):
    def filter(self, value):
        return value &gt; 0
class LessEquelToZero(FilterFunction):
    def filter(self, value):
        return value &lt;= 0
iteration = initial_stream.iterate(5000)
iteration_body = iteration.map(MinusOne())
feedback = iteration_body.filter(PositiveNumber())
iteration.close_with(feedback)
output = iteration_body.filter(LessEquelToZero())

将函数传递给Flink

某些算子操作需要用户定义的函数作为参数。所有函数都应该定义为派生自相关Flink函数的Python类。用户定义的函数被序列化并发送到TaskManagers以供执行。

class Filter(FilterFunction):
    def filter(self, value):
        return value > 5
data_stream.filter(Filter())

丰富的函数（.eg RichFilterFunction）允许定义（覆盖）可选算子操作：open＆close。用户可以使用这些函数进行初始化和清理。

class Tokenizer(RichMapFunction):
    def open(self, config):
        pass
    def close(self):
        pass
    def map(self, value):
        pass
data_stream.map(Tokenizer())

open在启动流式传输管道之前，Worker会调用该函数。close在流管道停止后，Worker调用该函数。

数据类型

Flink的Python Streaming API支持原始Python类型（int，float，bool，string），以及字节数组和用户定义的类。

class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age
class Tokenizer(MapFunction):
    def map(self, value):
        return (1, Person(*value))
data_stream.map(Tokenizer())

元组/列表

您可以将元组（或列表）用于复合类型。Python元组映射到Jython本机对应类型，这些类型由Python打包器薄层处理。

word_counts = env.from_elements(("hello", 1), ("world",2))
class Tokenizer(MapFunction):
    def map(self, value):
        return value[1]
counts = word_counts.map(Tokenizer())

数据源

数据源创建初始数据流，例如来自文件或集合。

基于文件的：

read_text_file(path) - 按行读取文件并将其作为字符串流返回。

基于集合：

from_elements(*args) - 从所有数据元创建数据流。
generate_sequence(from, to) - 并行生成给定间隔中的数字序列。

例子

env  = factory.get_execution_environment()
\# read text file from local files system
localLiens = env.read_text("file:///path/to/my/textfile")
\# read text file from a HDFS running at nnHost:nnPort
hdfsLines = env.read_text("hdfs://nnHost:nnPort/path/to/my/textfile")
\# create a set from some given elements
values = env.from_elements("Foo", "bar", "foobar", "fubar")
\# generate a number sequence
numbers = env.generate_sequence(1, 10000000)

数据接收

数据接收器使用DataStream并用于存储或返回它们：

write_as_text() - 按字符串顺序写入数据元。通过调用每个数据元的str（）方法获得字符串。
output()- 打印标准输出上每个数据元的str（）值。
write_to_socket() - 将DataStream作为字节数组写入套接字[host：port]。

可以将DataStream输入到多个算子操作。程序可以编写或打印数据流，同时对它们执行其他转换。

例子

标准数据接收方法：

 write DataStream to a file on the local file system
textData.write_as_text("file:///my/result/on/localFS")
 write DataStream to a file on a HDFS with a namenode running at nnHost:nnPort
textData.write_as_text("hdfs://nnHost:nnPort/my/result/on/localFS")
 write DataStream to a file and overwrite the file if it exists
textData.write_as_text("file:///my/result/on/localFS", WriteMode.OVERWRITE)
 this writes tuples in the text formatting "(a, b, c)", rather than as CSV lines
values.write_as_text("file:///path/to/the/result/file")

并行执行

本节介绍如何在Flink中配置程序的并行执行。Flink程序由多个任务（算子，数据源和接收器）组成。任务被分成几个并行实例以供执行，每个并行实例处理任务输入数据的子集。任务的并行实例数称为并行度或并行度（DOP）。

可以在不同级别的Flink中指定任务的并行度。

运行环境级别

Flink程序在运行环境的上下文中执行。运行环境为其执行的所有算子，数据源和数据接收器定义默认并行性。可以通过显式配置算子的并行性来覆盖运行环境并行性。

可以通过调用set_parallelism()方法来指定运行环境的默认并行性。要以并行方式执行WordCount示例程序的所有算子，数据源和数据接收器，请 3按如下方式设置运行环境的默认并行度：

env = factory.get_execution_environment()
env.set_parallelism(3)
text.flat_map(Tokenizer()) \
    .key_by(Selector()) \
    .time_window(milliseconds(30)) \
    .reduce(Sum()) \
    .print()
env.execute()

系统级别

可以通过设置parallelism.default属性来定义所有运行环境的系统范围默认并行度 ./conf/flink-conf.yaml。有关详细信息，请参阅配置文档

执行计划

要使用Flink运行计划，请转到Flink分发，然后从/ bin文件夹运行pyflink-stream.sh脚本。包含该计划的脚本必须作为第一个参数传递，然后是许多其他Python包，最后由-将被提供给脚本的其他参数分隔。

./bin/pyflink-stream.sh <Script>[ <pathToPackage1>[ <pathToPackageX]][ - <param1>[ <paramX>]]