Flink 使用介绍相关文档目录
前言
Flink提供了SQL client，有了它我们可以像Hive的beeline一样直接在控制台编写SQL并提交作业。完全不用使用Java/Scala编写Flink作业打包为jar提交。毫无疑问SQL client使用起来更为简单灵活。
如何使用
Flink SQL client支持运行在standalone集群和Yarn集群上。提交任务的命令有所不同。
SQL client运行于standalone模式：
./sql-client.sh embedded
注意。如果使用standalone模式运行，需要事先启动一个Flink standalone集群。方法如下
./start-cluster.sh
运行到Yarn集群上：
./sql-client.sh embedded -s yarn-session
大家可能会问：SQL slient怎么知道提交到哪个yarn app上？
指定yarn-session方式启动的时候会看到如下日志：
INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli [] - Found Yarn properties file under /tmp/.yarn-properties-root.
Flink每次启动yarn-session，都会创建一个/tmp/.yarn-properties-root文件(root为用户名，源码位于FlinkYarnSessionCli)。内容如下：
dynamicPropertiesString=
applicationID=application_xxxxxxxxxxxxx_xxxx
记录了最近一次提交的yarn session对应的application ID。
SQL client如果使用yarn session模式启动，会查找/tmp/.yarn-properties-root文件中指定的application id，将SQL提交到这个yarn session上运行。
场景示例
在本节我们使用Flink sql client，完成一个简单的场景：从Kafka中读取CSV数据将其写入MySQL。
准备数据
我们提前将准备好的测试数据灌入Kafka topic中，例如fludesc这个topic。使用如下方式查看Kafka topic中的数据：
./kafka-console-consumer.sh --topic fludesc --bootstrap-server broker1:9092,broker2:9092,broker3:9092 --from-beginning
部分样例数据如下：
511653962048,Zomfq,53,woman,532120,534.61,313020,cart,15926130785,UyxghCpKMD@huawei.com,2019-08-03
751653962048,Qvtil,27,man,532120,655.7,313023,cart,13257423096,cJfbNhRYow@163.com,2019-08-05
121653962048,Spdwh,35,woman,480071,97.35,313018,cart,18825789463,LkVYmpcWXC@qq.com,2019-08-05
871653962048,Fdhpc,18,man,650012,439.40,313012,cart,15059872140,sfzuPWvNEe@qq.com,2019-08-06
841653962048,Iqoyh,51,woman,152121,705.6,313012,buy,13646513897,jISbcYdxZO@126.com,2019-08-04
761653962048,Xgzhy,29,woman,480071,329.60,313013,cart,15069315824,NtTDRlAdeZ@qq.com,2019-08-04
接下来我们需要安装Flink和相关依赖。
安装Flink和依赖
从官网下载Flink解压到集群某节点任意目录（以Flink 1.13.2版本为例）。然后分析场景中需要用到哪些依赖。
项目中用到了CSV文件格式，读取Kafka和写入MySQL。我们需要找到下列依赖：
- flink-csv-1.13.2.jar: CSV格式支持
- flink-connector-jdbc_2.11-1.13.2.jar: 读写JDBC支持
- flink-connector-kafka_2.11-1.13.2.jar: 读写Kafka支持
- kafka-clients-2.4.1.jar: Kafka客户端
- mysql-connector-java-8.0.29.jar: MySQL JDBC驱动
将其放入Flink安装位置的lib目录下。
到这里Flink环境整理完毕。
启动Flink SQL client
首先启动Flink yarn-session。开启一个shell，执行：
export HADOOP_CLASSPATH=`hadoop classpath`
# 切换到具有访问HDFS和提交Yarn任务权限的用户
su hdfs
cd ${FLINK_HOME}/bin
./yarn-session.sh -s 1 -jm 1024 -tm 2048
启动yarn-session.sh的时候，可以添加-d参数使用detached模式，这样yarn session启动完毕之后shell就可以关闭。
然后启动Flink SQL client。
再开启一个shell，执行：
export HADOOP_CLASSPATH=`hadoop classpath`
# 切换到具有访问HDFS和提交Yarn任务权限的用户
su hdfs
cd ${FLINK_HOME}/bin
./sql-client.sh embedded -s yarn-session
看到Flink SQL Client的logo说明启动成功，可以编写SQL提交作业。
执行SQL提交作业
首先在MySQL中创建需要接收Kafka数据的表：
create database demo_db character set utf8mb4;
use demo_db;
create table fludesc (
id varchar(32),
use_rname varchar(32),
age int,
gender varchar(32),
goods_no varchar(32),
goods_price Float,
store_id int,
shopping_type varchar(32),
tel varchar(32),
email varchar(32),
shopping_date date
);
注：上面SQL的Schema需要和Kafka CSV数据对应。
然后我们在Flink SQL client中执行：
create table kafka_source (
id STRING,
use_rname STRING,
age integer,
gender STRING,
goods_no STRING,
goods_price Float,
store_id integer,
shopping_type STRING,
tel STRING,
email STRING,
shopping_date Date
) with (
'connector' = 'kafka',
'properties.bootstrap.servers' = 'broker1:9092,broker2:9092,broker3:9092',
'topic' = 'fludesc',
'scan.startup.mode' = 'earliest-offset',
'format' = 'csv',
'csv.ignore-parse-errors' = 'true'
);
CREATE TABLE mysql_sink (
id STRING,
use_rname STRING,
age integer,
gender STRING,
goods_no STRING,
goods_price Float,
store_id integer,
shopping_type STRING,
tel STRING,
email STRING,
shopping_date Date
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://mysql_server:3306/demo_db',
'table-name' = 'fludesc',
'username' = 'root',
'password' = 'password'
);
insert into mysql_sink select * from kafka_source;
参数解析：
- properties.bootstrap.servers：Kafka broker地址和端口号。
- topic: 数据源Kafka topic 名称。
- scan.startup.mode: 例子中为从Kafka topic中从头开始消费。
- format: 使用的数据格式，样例中为CSV。
- url: 连接MySQL的JDBC连接串。
- table-name: MySQL表名。
- username: MySQL用户名。
- password: MySQL密码。
- csv.ignore-parse-errors: 忽略解析错误。启用此配置可以容忍脏数据。
如果全过程没有问题，可进入MySQL控制台执行：
use demo_db;
select * from fludesc;
可以查看到Kafka中的数据已写入到MySQL。
数据类型
Flink SQL和MySQL数据类型在SQL中写法是不一样的。具体写法和对照关系可参考：
Flink SQL数据类型：Data Types | Apache Flink
MySQL数据类型：MySQL 数据类型 | 菜鸟教程 (runoob.com)
常见错误
找不到Hadoop相关依赖
启动yarn-session的时候出现：
Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR or HADOOP_CLASSPATH was set.
Setting HBASE_CONF_DIR=/etc/hbase/conf because no HBASE_CONF_DIR was set.
Error: A JNI error has occurred, please check your installation and try again
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/exceptions/YarnException
at java.lang.Class.getDeclaredMethods0(Native Method)
at java.lang.Class.privateGetDeclaredMethods(Class.java:2701)
at java.lang.Class.privateGetMethodRecursive(Class.java:3048)
at java.lang.Class.getMethod0(Class.java:3018)
at java.lang.Class.getMethod(Class.java:1784)
at sun.launcher.LauncherHelper.validateMainClass(LauncherHelper.java:544)
at sun.launcher.LauncherHelper.checkAndLoadMain(LauncherHelper.java:526)
Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.yarn.exceptions.YarnException
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 7 more
说明Flink找不到HADOOP的相关配置和依赖。需要提前执行：
export HADOOP_CLASSPATH=`hadoop classpath`
SQL client 提交任务的时候连接拒绝
[INFO] SUbmitting SQL update statement to the cluster...
[ERROR] Could not execute SQL statement. Reason:
java.net.ConnectException: Connection refused
这个问题报错比较笼统。真实的原因是sql-client无法连接到Flink集群的job manager。
如果使用standalone模式，需要执行./start-cluster.sh启动一个standalone集群。
如果使用Yarn session模式，则需要：
- 启动sql client之前需要export HADOOP_CLASSPATH环境变量。
- 提交yarn session和启动sql client需要使用同一个用户，否则会找不到yarn session对应的application id。
- 确保当前机器的Yarn客户端配置无问题。可通过执行yarn命令是否能正常返回集群信息确认。
网友评论