flink1.12.3+python3.7+pyflink实时消费kafka

一、环境介绍

二、思路梳理

三、启动zookeeper、kafka、flink集群（本地）等步骤省略

四、数据生成器（这个测试的时候用的python2，问题不大）

五、pyflink实时读流

六、提交作业

七、一些异常和报错

一、环境介绍

centos7、anaconda的虚拟环境python3.7、官网原生flink1.12.3、官网原生kafka_2.12-2.7.0、官网原生apache-zookeeper-3.6.3-bin

开始想用python2来搞，但是问题比较多，参看官网，python2.7只能支持到flink1.9.3，后面的tableAPI没有什么支持。

二、思路梳理

写一个数据生成器往kafka里面生产数据——>pyflink实时消费kafka的流数据

三、启动zookeeper、kafka、flink集群（本地）等步骤省略

四、数据生成器（这个测试的时候用的python2，问题不大）

# --coding=utf8--
import kafka
# 循环向kafka写数
import datetime
import time


kp=kafka.KafkaProducer(bootstrap_servers="localhost:9092")
for i in range(0,100):
    kp.send(topic="haha",value="test_"+datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S"))
    time.sleep(1)
kp.flush()
kp.close()

五、pyflink实时读流

pip install apache-flink==1.12.0

此处参考了一个前辈的代码，这里根据个人需要测试，进行了修改，原文在这：

pyflink消费kafka-connect-jdbc消息(带schema) - 知乎

#--coding=utf8--

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, TableConfig

# 1.创建执行环境
s_env = StreamExecutionEnvironment.get_execution_environment()
s_env.set_parallelism(1)
st_env = StreamTableEnvironment.create(s_env, TableConfig())
st_env.get_config().set_python_executable("python3")


# 2.利用实时流建表，format的raw应该表示的是非格式化的原生内容（raw的单词意思）
ddlKafkaConn = """
create table sourceKafkaConn(
    `CREATE_TIME` STRING   comment '行数据'
)
with(
    'connector' = 'kafka',
    'topic' = 'haha',        
    'properties.bootstrap.servers' = 'localhost:9092',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'raw'
)
"""
# 'connector.startup-mode' = 'earliest-offset 表示读取最早的消息 | latest-offset 表示读取消息队列中最新的消息',
st_env.execute_sql(ddlKafkaConn)

# 3.创建一个sink
sinkPrint = '''
    CREATE TABLE sinkPrint WITH ('connector' = 'print')
    LIKE sourceKafkaConn (EXCLUDING ALL)
'''
st_env.execute_sql(sinkPrint)

# 4.查询实时流中的数据并插入到sink中
st_env.sql_query("select `CREATE_TIME` as payload from sourceKafkaConn") \
    .execute_insert("sinkPrint")

六、提交作业

这里的kafka-clients-2.7.0.jar是从kafka的libs文件夹cp过来的，后来因为一些报错，又从maven的官网手动下载了一个kafka-clients-2.4.1.jar。但是当时的报错应该不是这里引起的，有兴趣的朋友可以都试一下。

bash flink run -py job.py -j kafka-clients-2.7.0.jar

可以从flink的页面上的taskmanager的stdout标准输出找到我们读到的数据，这里能打印应该是指定sink是print的。

七、一些异常和报错

1.NoClassDefFoundError: org/apache/kafka/clients/consumer/ConsumerRecord

这就是需要在提交作业的时候指定kafka-clients的jar包，如果没有则需在flink的lib文件夹中添加一个，同时在代码中的connector' = 'kafka',这里可以尝试写别的值，如果值不符合规范，则会报错告诉你可选项，或者参考相关文档查看选项。

2.【Flink】JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy

这是参考另一个前辈的文档，将flink默认配置1改大些：

taskmanager.numberOfTaskSlots: 2

parallelism.default: 2

参考连接：flink shell的local模式(benv与senv的使用+处理报错的解决方案)_微电子学与固体电子学-CSDN博客

里面的-附录-报错处理2