1. 前言
Flink是批流一体化的数据处理框架，性能卓越，诸多大厂都在使用。
由于时间原因，本篇文章只简单了解一下，后续会深入分析，以及经典case分享。
- Flink开源社区的视频教程，很赞
https://space.bilibili.com/33807709
2. 安装pyflink
- 安装虚拟环境和pyflink
$ conda create -n py36 python=3.6
$ conda activate py36
$ conda install Cython
$ python -m pip install apache-flink==1.10.1
- runtime需要jdk，下载jdk 10后解压，配置env
export JAVA_HOME=/home/xxx/java/jdk-10.0.2
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
- flink 1.10 最高支持python3.7，暂不支持py38
- 通过clone repo的方式，可以安装最新版本的pyflink
3. run server
- 创建一个docker-compose，启动flink单机cluster
$ cat docker-compose.yml
version: "2.1"
services:
jobmanager:
image: ${FLINK_DOCKER_IMAGE_NAME:-flink}
expose:
- "6123"
ports:
- "8081:8081"
command: jobmanager
environment:
- JOB_MANAGER_RPC_ADDRESS=jobmanager
taskmanager:
image: ${FLINK_DOCKER_IMAGE_NAME:-flink}
expose:
- "6121"
- "6122"
depends_on:
- jobmanager
command: taskmanager
links:
- "jobmanager:jobmanager"
environment:
- JOB_MANAGER_RPC_ADDRESS=jobmanager
- 启动flink，确认服务ok
$ docker-compose up -d
$ docker-compose scale taskmanager=3
$ docker-compose ps
Name Command State Ports
-----------------------------------------------------------------------------------------------------
flink_jobmanager_1 /docker-entrypoint.sh jobm ... Up 6123/tcp, 0.0.0.0:8081->8081/tcp
flink_taskmanager_1 /docker-entrypoint.sh task ... Up 6121/tcp, 6122/tcp, 6123/tcp, 8081/tcp
flink_taskmanager_2 /docker-entrypoint.sh task ... Up 6121/tcp, 6122/tcp, 6123/tcp, 8081/tcp
flink_taskmanager_3 /docker-entrypoint.sh task ... Up 6121/tcp, 6122/tcp, 6123/tcp, 8081/tcp
- flink dashboard
http://10.30.4.21:8081/#/overview
image.png
4. Job开发与测试
test example (pyflink本地启动了mini cluster，未使用我们启动的flink)
cd到pyflink的安装目录，参考： ~/soft/miniconda3/envs/py36/lib/python3.6/site-packages/pyflink
执行word_count.py（大数据版hello-world）
$ python examples/python/table/batch/word_count.py
pyflink shell streaming api
$ ./bin/pyflink-shell.sh remote localhost 8081
# 输入streaming的demo代码
import tempfile
import os
import shutil
sink_path = tempfile.gettempdir() + '/streaming.csv'
if os.path.exists(sink_path):
if os.path.isfile(sink_path):
os.remove(sink_path)
else:
shutil.rmtree(sink_path)
s_env.set_parallelism(1)
t = st_env.from_elements([(1, 'hi', 'hello'), (2, 'hi', 'hello')], ['a', 'b', 'c'])
st_env.connect(FileSystem().path(sink_path)) \
.with_format(OldCsv()
.field_delimiter(',')
.field("a", DataTypes.BIGINT())
.field("b", DataTypes.STRING())
.field("c", DataTypes.STRING())) \
.with_schema(Schema()
.field("a", DataTypes.BIGINT())
.field("b", DataTypes.STRING())
.field("c", DataTypes.STRING())) \
.register_table_sink("stream_sink")
t.select("a + 1, b, c").insert_into("stream_sink")
st_env.execute("stream_job")
- exec到task container，查看结果文件，结果ok
# cat /tmp/streaming.csv
2,hi,hello
3,hi,hello
-
dashboard查看job执行情况
image.png
其他提交job方式
通过flink提交py脚本，或者，直接提交flink sql
$ ./bin/flink run -py examples/python/table/batch/word_count.py
