Pyflink 的安装和 windows 开发环境配置_pyflink window

在分布式大数据计算引擎这个领域，现在最常用的Apache Spark 早已支持 python语言编写，而且对ML（机器学习）和DM（数据挖掘）也都有api的支持，而作为第三代计算引擎的flink， 从 1.9.0 版本开始增加了对 Python 的支持（PyFlink），在Flink 1.10 中，PyFlink 添加了对 Python UDFs（自定义函数） 的支持，可以在 Table API/SQL 中注册并使用自定义函数，而从Flink 1.11开始还支持在 Windows 上本地运行 PyFlink 作业，所以可以在 Windows 上开发和调试 PyFlink 作业

Pyflink的安装

Pyflink的安装十分之简单

• 首先先看看自己系统的python版本（PyFlink 需要 Python 版本（3.6、3.7 或 3.8））

$ python3 --version
# the version printed here must be 3.6, 3.7 or 3.8
1
2

• 环境配置
  由于系统可能包含多个 Python 版本，因此也包含多个 Python 二进制可执行文件。运行以下 ls命令来找出系统中可用的 Python 二进制可执行文件：

$ ls /usr/bin/python*
1

• 选择软链接python指向您的python3解释器

ln -s /usr/bin/python3 python
1

• 安装Pyflink
  由于pyflink还在持续火热的更新之中每个版本的变化较大，所以无脑安装最新版就行（更新此文章时最新版为apache-flink1.13.2）

$ python3 -m pip install apache-flink
1

windows开发环境配置

这里我们选择 Pycharm IDE 来进行windows Pyflink 开发

• 首先配置Pytthon 虚拟环境
  配置路径PyCharm -> Preferences -> Project Interpreter

记住选择的python版本一定要是3.6、3.7 或 3.8

• 新建一个项目，环境选择我们刚刚配置的python 虚拟环境

• 安装Pyflink
  进入到终端界面
  
  先看看 python 版本
  

  然后安装 Pyflink
  
  最终完成之后你可以在 site-packages下面找的 pyflink目录，如下
  

• Hello World 示例

新建一个.py文件输入以下代码

#!/usr/bin/env python38
#-*- coding:utf-8 -*-
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import EnvironmentSettings, StreamTableEnvironment

def hello_world():
    """
    从随机Source读取数据，然后直接利用PrintSink输出。
    """
    settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(stream_execution_environment=env, environment_settings=settings)
    source_ddl = """
                    CREATE TABLE random_source (
                        f_sequence INT,
                        f_random INT,
                        f_random_str STRING
                    ) WITH (
                        'connector' = 'datagen',
                        'rows-per-second'='5',
                        'fields.f_sequence.kind'='sequence',
                        'fields.f_sequence.start'='1',
                        'fields.f_sequence.end'='1000',
                        'fields.f_random.min'='1',
                        'fields.f_random.max'='1000',
                        'fields.f_random_str.length'='10'
                    )
                    """

    sink_ddl = """
                  CREATE TABLE print_sink (
                    f_sequence INT,
                    f_random INT,
                    f_random_str STRING 
                ) WITH (
                  'connector' = 'print'
                )
        """

    # 注册source和sink
    t_env.execute_sql(source_ddl)
    t_env.execute_sql(sink_ddl)

    # 数据提取
    tab = t_env.from_path("random_source")
    # 这里我们暂时先使用 标注了 deprecated 的API, 因为新的异步提交测试有待改进...
    tab.execute_insert("print_sink").wait()
    # 执行作业
    t_env.execute_sql("Flink Hello World")



if __name__ == '__main__':
    hello_world()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

结果如下

欢迎交流学习

个人博客

掘金主页