PyFlink 开发环境利器:Zeppelin Notebook

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 在 Zeppelin notebook 里利用 Conda 来创建 Python env 自动部署到 Yarn 集群中。

PyFlink 作为 Flink 的 Python 语言入口,其 Python 语言的确很简单易学,但是 PyFlink 的开发环境却不容易搭建,稍有不慎,PyFlink 环境就会乱掉,而且很难排查原因。今天给大家介绍一款能够帮你解决这些问题的 PyFlink 开发环境利器:Zeppelin Notebook。主要内容为:

  1. 准备工作
  2. 搭建 PyFlink 环境
  3. 总结与未来

GitHub 地址
https://github.com/apache/flink
欢迎大家给 Flink 点赞送 star~

也许你早就听说过 Zeppelin,但是之前的文章都偏重讲述如何在 Zeppelin 里开发 Flink SQL,今天则来介绍下如何在 Zeppelin 里高效的开发 PyFlink Job,特别是解决 PyFlink 的环境问题。

一句来总结这篇文章的主题,就是在 Zeppelin notebook 里利用 Conda 来创建 Python env 自动部署到 Yarn 集群中,你无需手动在集群上去安装任何 PyFlink 的包,并且你可以在一个 Yarn 集群里同时使用互相隔离的多个版本的 PyFlink。最后你能看到的效果就是这样:

1. 能够在 PyFlink 客户端使用第三方 Python 库,比如 matplotlib:

img

2. 可以在 PyFlink UDF 里使用第三方 Python 库,如:

img

接下来看看如何来实现。

一、准备工作

Step 1.

准备好最新版本的 Zeppelin 的搭建,这个就不在这边展开了,如果有问题可以加入 Flink on Zeppelin 钉钉群 (34517043) 咨询。另外需要注意的是,Zeppelin 部署集群需要是 Linux,如果是 Mac 的话,会导致在 Mac 机器上打的 Conda 环境无法在 Yarn 集群里使用 (因为 Conda 包在不同系统间是不兼容的)。

Step 2.

下载 Flink 1.13, 需要注意的是,本文的功能只能用在 Flink 1.13 以上版本,然后:

  • flink-Python-*.jar 这个 jar 包 copy 到 Flink 的 lib 文件夹下;
  • opt/Python 这个文件夹 copy 到 Flink 的 lib 文件夹下。

Step 3.

安装以下软件 (这些软件是用于创建 Conda env 的):

二、搭建 PyFlink 环境

接下来就可以在 Zeppelin 里搭建并且使用 PyFlink 了。

Step 1. 制作 JobManager 上的 PyFlink Conda 环境

因为 Zeppelin 天生支持 Shell,所以可以在 Zeppelin 里用 Shell 来制作 PyFlink 环境。注意这里的 Python 第三方包是在 PyFlink 客户端 (JobManager) 需要的包,比如 Matplotlib 这些,并且确保至少安装了下面这些包:

  • 某个版本的 Python (这里用的是 3.7)
  • apache-flink (这里用的是 1.13.1)
  • jupyter,grpcio,protobuf (这三个包是 Zeppelin 需要的)

剩下的包可以根据需要来指定:

%sh

# make sure you have conda and momba installed.
# install miniconda: https://docs.conda.io/en/latest/miniconda.html
# install mamba: https://github.com/mamba-org/mamba

echo "name: pyflink_env
channels:
  - conda-forge
  - defaults
dependencies:
  - Python=3.7
  - pip
  - pip:
    - apache-flink==1.13.1
  - jupyter
  - grpcio
  - protobuf
  - matplotlib
  - pandasql
  - pandas
  - scipy
  - seaborn
  - plotnine
 " > pyflink_env.yml
    
mamba env remove -n pyflink_env
mamba env create -f pyflink_env.yml

运行下面的代码打包 PyFlink 的 Conda 环境并且上传到 HDFS (注意这里打包出来的文件格式是 tar.gz):

%sh

rm -rf pyflink_env.tar.gz
conda pack --ignore-missing-files -n pyflink_env -o pyflink_env.tar.gz

hadoop fs -rmr /tmp/pyflink_env.tar.gz
hadoop fs -put pyflink_env.tar.gz /tmp
# The Python conda tar should be public accessible, so need to change permission here.
hadoop fs -chmod 644 /tmp/pyflink_env.tar.gz

Step 2. 制作 TaskManager 上的 PyFlink Conda 环境

运行下面的代码来创建 TaskManager 上的 PyFlink Conda 环境,TaskManager 上的 PyFlink 环境至少包含以下 2 个包:

  • 某个版本的 Python (这里用的是 3.7)
  • apache-flink (这里用的是 1.13.1)

剩下的包是 Python UDF 需要依赖的包,比如这里指定了 pandas:

echo "name: pyflink_tm_env
channels:
  - conda-forge
  - defaults
dependencies:
  - Python=3.7
  - pip
  - pip:
    - apache-flink==1.13.1
  - pandas
 " > pyflink_tm_env.yml
    
mamba env remove -n pyflink_tm_env
mamba env create -f pyflink_tm_env.yml

运行下面的代码打包 PyFlink 的 conda 环境并且上传到 HDFS (注意这里使用的是 zip 格式)

%sh

rm -rf pyflink_tm_env.zip
conda pack --ignore-missing-files --zip-symlinks -n pyflink_tm_env -o pyflink_tm_env.zip

hadoop fs -rmr /tmp/pyflink_tm_env.zip
hadoop fs -put pyflink_tm_env.zip /tmp
# The Python conda tar should be public accessible, so need to change permission here.
hadoop fs -chmod 644 /tmp/pyflink_tm_env.zip

Step 3. 在 PyFlink 中使用 Conda 环境

接下来就可以在 Zeppelin 中使用上面创建的 Conda 环境了,首先需要在 Zeppelin 里配置 Flink,主要配置的选项有:

  • flink.execution.mode 为 yarn-application, 本文所讲的方法只适用于 yarn-application 模式;
  • 指定 yarn.ship-archives,zeppelin.pyflink.Python 以及 zeppelin.interpreter.conda.env.name 来配置 JobManager 侧的 PyFlink Conda 环境;
  • 指定 Python.archives 以及 Python.executable 来指定 TaskManager 侧的 PyFlink Conda 环境;
  • 指定其他可选的 Flink 配置,比如这里的 flink.jm.memory 和 flink.tm.memory。
%flink.conf


flink.execution.mode yarn-application

yarn.ship-archives /mnt/disk1/jzhang/zeppelin/pyflink_env.tar.gz
zeppelin.pyflink.Python pyflink_env.tar.gz/bin/Python
zeppelin.interpreter.conda.env.name pyflink_env.tar.gz

Python.archives hdfs:///tmp/pyflink_tm_env.zip
Python.executable  pyflink_tm_env.zip/bin/Python3.7

flink.jm.memory 2048
flink.tm.memory 2048

接下来就可以如一开始所说的那样在 Zeppelin 里使用 PyFlink 以及指定的 Conda 环境了。有 2 种场景:

  • 下面的例子里,可以在 PyFlink 客户端 (JobManager 侧) 使用上面创建的 JobManager 侧的 Conda 环境,比如下边使用了 Matplotlib。

    img

  • 下面的例子是在 PyFlink UDF 里使用上面创建的 TaskManager 侧 Conda 环境里的库,比如下面在 UDF 里使用 Pandas。

    img

三、总结与未来

本文内容就是在 Zeppelin notebook 里利用 Conda 来创建 Python env 自动部署到 Yarn 集群中,无需手动在集群上去安装任何 Pyflink 的包,并且可以在一个 Yarn 集群里同时使用多个版本的 PyFlink。

每个 PyFlink 的环境都是隔离的,而且可以随时定制更改 Conda 环境。可以下载下面这个 note 并导入到 Zeppelin,就可以复现今天讲的内容:http://23.254.161.240/#/notebook/2G8N1WTTS

此外还有很多可以改进的地方:

  • 目前我们需要创建 2 个 conda env ,原因是 Zeppelin 支持 tar.gz 格式,而 Flink 只支持 zip 格式。等后期两边统一之后,只要创建一个 conda env 就可以;
  • apache-flink 现在包含了 Flink 的 jar 包,这就导致打出来的 conda env 特别大,yarn container 在初始化的时候耗时会比较长,这个需要 Flink 社区提供一个轻量级的 Python 包 (不包含 Flink jar 包),就可以大大减小 conda env 的大小。

希望了解更多 Flink on Zeppelin 使用的同学可以加入下面的钉钉群来讨论。

img


第三届 Apache Flink 极客挑战赛报名开始!
30 万奖金等你来!

伴随着海量数据的冲击,数据处理分析能力在业务中的价值与日俱增,各行各业对于数据处理时效性的探索也在不断深入,作为主打实时计算的计算引擎 - Apache Flink 应运而生。

为给行业带来更多实时计算赋能实践的思路,鼓励广大热爱技术的开发者加深对 Flink 的掌握,Apache Flink 社区联手阿里云、英特尔、阿里巴巴人工智能治理与可持续发展实验室 (AAIG)、Occlum 联合举办 "第三届 Apache Flink 极客挑战赛暨 AAIG CUP" 活动,即日起正式启动。

👉 点击了解更多赛事信息 👈

img


更多 Flink 相关技术问题,可扫码加入社区钉钉交流群
第一时间获取最新技术文章和社区动态,请关注公众号~

image.png

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算Flink版现开启活动:
99 元试用 实时计算Flink版(包年包月、10CU)即有机会获得 Flink 独家定制T恤;另包 3 个月及以上还有 85 折优惠!
了解活动详情:https://www.aliyun.com/product/bigdata/sc

image.png

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
流计算 资源调度 Java
Apache Flink 零基础入门(二):开发环境搭建和应用的配置、部署及运行
本文主要面向于初次接触 Flink、或者对 Flink 有了解但是没有实际操作过的同学。希望帮助大家更顺利地上手使用 Flink,并着手相关开发调试工作。
Apache Flink 零基础入门(二):开发环境搭建和应用的配置、部署及运行
|
6月前
|
数据采集 机器学习/深度学习 数据可视化
Jupyter Notebooks和IPython是交互式数据分析利器,提升效率。Jupyter是开源Web应用
【7月更文挑战第5天】Jupyter Notebooks和IPython是交互式数据分析利器,提升效率。Jupyter是开源Web应用,支持多语言,结合IPython的交互式解释器,便于编程和科学计算。两者提供即时反馈、丰富库支持、跨语言功能及协作共享。基本流程包括:数据导入(使用Pandas)、预处理、分析(借助Pandas、NumPy、Matplotlib)、模型训练(如随机森林)和评估。
59 0
|
8月前
|
SQL Java 数据处理
实时计算 Flink版产品使用合集之怎么热加载Java和Python的UDF
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
102 1
|
8月前
|
JSON 数据可视化 数据挖掘
适合数据分析的ide---Jupyter Notebook的安装使用
适合数据分析的ide---Jupyter Notebook的安装使用
134 2
|
8月前
|
分布式计算 Hadoop 关系型数据库
Sqoop入门指南:安装和配置
Sqoop入门指南:安装和配置
|
8月前
|
SQL 缓存 分布式计算
Apache Zeppelin系列教程第九篇——Zeppelin NoteBook数据缓存
Apache Zeppelin系列教程第九篇——Zeppelin NoteBook数据缓存
163 0
|
分布式计算 资源调度 Spark
Zeppelin_解释器配置| 学习笔记
快速学习 Zeppelin_解释器配置
Zeppelin_解释器配置| 学习笔记
|
机器学习/深度学习 人工智能 分布式计算
Apache Flink 说道系列 - PyFlink集成Pandas(1+1 != 2)
# 开篇说道 说道聊什么,聊阿里人熟知的“因为相信,所以简单”!这是每个人内心所神往的日常工作生活!这看似 简单的言语,透彻的道理,虽被大众所认可,但人们对其执行的能力却大相径庭。“因为相信,所以简单” 不是对我们的约束,而是为我们寻求快乐指明了方向...永远做别人成功路上的基石...当TA人踏上人生巅峰,基石自具人生高度! ![image.png](https://ata2-img
3454 0
Apache Flink 说道系列 - PyFlink集成Pandas(1+1 != 2)
|
Java API 流计算
|
SQL 存储 Java