《Flink官方文档》Python 编程指南测试版（一）

Flink中的分析程序实现了对数据集的某些操作 (例如，数据过滤，映射，合并，分组)。这些数据最初来源于特定的数据源(例如来自于读文件或数据集合)。操作执行的结果通过数据池以写入数据到(分布式)文件系统或标准输出(例如命令行终端)的形式返回。Flink程序可以运行在不同的环境中，既能够独立运行，也可以嵌入到其他程序中运行。程序可以运行在本地的JVM上，也可以运行在服务器集群中。

为了创建你自己的Flink程序，我们鼓励你从program skeleton（程序框架）开始，并逐渐增加你自己的transformations（变化）。以下是更多的用法和高级特性的索引。

• Example Program
• Program Skeleton
• Project setup
• Lazy Evaluation
• Transformations
• Specifying Keys
• Passing Functions to Flink
• Data Types
• Data Sources
• Data Sinks
• Broadcast Variables
• Parallel Execution
  • Execution Environment Level
  • System Level
• Executing Plans

示例程序

以下程序是一段完整可运行的WordCount示例程序。你可以复制粘贴这些代码并在本地运行。

from flink.plan.Environment import get_environment
from flink.functions.GroupReduceFunction import GroupReduceFunction

class Adder(GroupReduceFunction):
  def reduce(self, iterator, collector):
    count, word = iterator.next()
    count += sum([x[0] for x in iterator])
    collector.collect((count, word))

env = get_environment()
data = env.from_elements("Who's there?",
 "I think I hear them. Stand, ho! Who's there?")

data \
  .flat_map(lambda x, c: [(1, word) for word in x.lower().split()]) \
  .group_by(1) \
  .reduce_group(Adder(), combinable=True) \
  .output()

env.execute(local=True)

程序框架

从示例程序可以看出，Flink程序看起来就像普通的python程序一样。每个程序都包含相同的基本组成部分：

1.获取一个运行环境
2.加载/创建初始数据
3.指定对这些数据的操作
4.指定计算结果的存放位置
5.运行程序

接下来，我们将对每个步骤给出概述，更多细节可以参考与之对应的小节。
Environment（运行环境）是所有Flink程序的基础。你可以通过调用Environment类中的一些静态方法来建立一个环境:

get_environment()

运行环境可通过多种读文件的方式来指定数据源。如果是简单的按行读取文本文件，你可以采用:

env = get_environment()
text = env.read_text("file:///path/to/file")

这样，你就获得了可以进行操作（apply transformations）的数据集。关于数据源和输入格式的更多信息，请参考 Data Sources。
一旦你获得了一个数据集DataSet，你就可以通过transformations来创建一个新的数据集，并把它写入到文件，再次transform，或者与其他数据集相结合。你可以通过对数据集调用自己个性化定制的函数来进行数据操作。例如，一个类似这样的数据映射操作：

data.map(lambda x: x*2)

这将会创建一个新的数据集，其中的每个数据都是原来数据集中的2倍。若要获取关于所有transformations的更多信息，及所有数据操作的列表，请参考Transformations。

当你需要将所获得的数据集写入到磁盘时，调用下面三种函数的其中一个即可。

data.write_text("<file-path>", WriteMode=Constants.NO_OVERWRITE)
write_csv("<file-path>", line_delimiter='\n', field_delimiter=',', write_mode=Constants.NO_OVERWRITE)
output()

其中，最后一种方法仅适用于在本机上进行开发/调试，它会将数据集的内容输出到标准输出。（请注意，当函数在集群上运行时，结果将会输出到整个集群节点的标准输出流，即输出到workers的.out文件。）前两种方法，能够将数据集写入到对应的文件中。关于写入到文件的更多信息，请参考Data Sinks。

当你设计好了程序之后，你需要在环境中执行execute命令来运行程序。可以选择在本机运行，也可以提交到集群运行，这取决于Flink的创建方式。你可以通过设置execute(local=True)强制程序在本机运行。

创建项目

除了搭建好Flink运行环境，就无需进行其他准备工作了。Python包可以从你的Flink版本对应的/resource文件夹找到。在执行工作任务时，Flink 包，plan包和optional包均可以通过HDFS自动分发。

Python API已经在安装了Python2.7或3.4的Linux/Windows系统上测试过。

默认情况下，Flink通过调用”python”或”python3″来启动python进程，这取决于使用了哪种启动脚本。通过在 flink-conf.yaml 中设置 “python.binary.python[2/3]”对应的值，来设定你所需要的启动方式。

延迟(惰性)求值

所有的Flink程序都是延迟执行的。当程序的主函数执行时，数据的载入和操作并没有在当时发生。与此相反，每一个被创建出来的操作都被加入到程序的计划中。当程序环境中的某个对象调用了execute()函数时，这些操作才会被真正的执行。不论该程序是在本地运行还是集群上运行。

延迟求值能够让你建立复杂的程序，并在Flink上以一个整体的计划单元来运行。

数据变换

数据变换（Data transformations）可以将一个或多个数据集映射为一个新的数据集。程序能够将多种变换结合到一起来进行复杂的整合变换。

该小节将概述各种可以实现的数据变换。transformations documentation数据变换文档中，有关于所有数据变换和示例的全面介绍。

data.map(lambda x: x * 2)

data.flat_map(
  lambda x,c: [(1,word) for word in line.lower().split() for line 
in x])

data.map_partition(lambda x,c: [value * 2 for value in x])

data.filter(lambda x: x > 1000)

data.reduce(lambda x,y : x + y)

lass Adder(GroupReduceFunction):
  def reduce(self, iterator, collector):
    count, word = iterator.next()
    count += sum([x[0] for x in iterator)      
    collector.collect((count, word))

data.reduce_group(Adder())

# This code finds the sum of all of the values in the first field
 and the maximum of all of the values in the second field
data.aggregate(Aggregation.Sum, 0).and_agg(Aggregation.Max, 1)

# min(), max(), and sum() syntactic sugar functions are also available
data.sum(0).and_agg(Aggregation.Max, 1)

# In this case tuple fields are used as keys.
# "0" is the join field on the first tuple
# "1" is the join field on the second tuple.
result = input1.join(input2).where(0).equal_to(1)

data1.co_group(data2).where(0).equal_to(1)

result = data1.cross(data2)

data.union(data2)

data.zip_with_index()

指定Keys

一些变换（例如Join和CoGroup），需要在进行变换前，为作为输入参数的数据集指定一个关键字，而另一些变换（例如Reduce和GroupReduce），则允许在变换操作之前，对数据集根据某个关键字进行分组。

数据集可通过如下方式分组

reduced = data \
  .group_by(<define key here>) \
  .reduce_group(<do something>)

Flink中的数据模型并不是基于键-值对。你无需将数据集整理为keys和values的形式。键是”虚拟的”：它们被定义为在真实数据之上，引导分组操作的函数。

转载自 并发编程网 - ifeve.com