接着上一个还是   FusionInsight/client/Flink的版本[ERROR] Flink SQL Client JAR file 'flink-sql-client*.jar'这个报错是?./bin/sql-client.sh embedded[ERROR] Flink SQL Client JAR file 'flink-sql-client*.jar' neither found in classpath nor /opt directory should be located in /xxxx/xxxFusionInsightxxx/xxclient/Flink/flink/opt.还要额外放入jar包.  才能跑起来吗?   

  华为版本 fi的 flink    直接运行start-cluster.sh 报错了, 是不是不用直接启动?       ./start-cluster.shNo masters file. Please specify masters in 'conf/masters'.默认的路径下载有flink配置  /srv/BigData/data1/FusionInsight/client/Flink/flink/bin    ,启动./start-cluster.sh 报错了  No masters file. Please specify masters in 'conf/masters'.求支援!

# 视频在本帖二楼 ## 一、Flink简介 Flink是一个批处理和流处理结合的统一计算框架，其核心是一个提供了数据分发以及并行化计算的流数据处理引擎。它的最大亮点是流处理，是业界最顶级的开源流处理引擎。 ## 二、Flink客户端准备，可参考MRS二次开发(15/27)Flink构造DataStream样例,有详细的视频介绍 **Kerberos认证** 从服务器上下载用户keytab，并将keytab放到Flink客户端所在主机的某个文件夹下。 修改客户端配置文件/opt/client/Flink/flink/conf/flink-conf.yaml内容： ``` keytab路径。例如：security.kerberos.login.keytab: /opt/testclient/flinkuser.keytab principal名。例如：security.kerberos.login.principal: flinkuser 对于HA模式，如果配置了ZooKeeper，还需要设置ZK kerberos认证相关的配置。配置如下： zookeeper.sasl.disable: false security.kerberos.login.contexts: Client 如果用户对于Kafka client和Kafka broker之间也需要做kerberos认证，配置如下： security.kerberos.login.contexts: Client,KafkaClient ``` **Security Cookie认证** 1. 用户需要获取SSL证书，放置到Flink客户端中。 参考 拷贝其中代码生成generate_keystore.sh脚本，放置在Flink客户端的bin目录下。 在客户端目录下执行source bigdata_env。 执行命令“sh generate_keystore.sh ”即可，例如“sh generate_keystore.sh 123456"，会在Flink客户端的conf目录下生成flink.keystore，flink.truststore文件。 在Flink客户端同级目录下新建ssl目录，例如“/opt/testclient/Flink/flink/ssl”，将生成的flink.keystore，flink.truststore文件拷贝到ssl目录中。 1. 获取证书后，在Flink客户端的conf目录下配置文件“flink-conf.yaml”中将以下配置项进行赋值。 将配置项“security.ssl.keystore”设置为keystore文件的相对路径，例如“ssl/flink.keystore” 将配置项“security.ssl.truststore”设置为truststore文件的相对路径，例如“ssl/flink.truststore” 将配置项“security.cookie”设置为一串密码，该密码可以是一串随机规则密码，可以取默认 使用Manager明文加密API进行获取密文：curl -k -i -u 用户名:密码 -X POST -HContent-type:application/json -d '{"plainText":""}' ''；其中要与签发证书时使用的密码一致，x.x.x.x为集群Manager的浮动IP。 将配置项“security.ssl.key-password”、“security.ssl.keystore-password”和“security.ssl.truststore-password”设置为如上指令获取到的密文。 2. 打开“Security Cookie”开关，配置“security.enable: true”，查看“security cookie”是否已配置成功。 #### 加密传输 配置SSL传输，用户主要在客户端的“flink-conf.yaml”文件中做如下配置： 1. 打开SSL开关和设置SSL加密算法,配置如下： security.ssl.enabled：true akka.ssl.enabled: true blob.service.ssl.enabled: true taskmanager.data.ssl.enabled: true security.ssl.algorithms: TLS_DHE_RSA_WITH_AES_128_CBC_SHA256,TLS_DHE_DSS_WITH_AES_128_CBC_SHA256 2. 配置keystore或truststore文件路径为相对路径时，Flink Client执行命令的目录需要可以直接访问该相对路径。 在Flink的CLI yarn-session.sh命令中增加“-t”选项来传输keystore和truststore文件到各个执行节点。例如“./bin/yarn-session.sh **-t** ssl/ ” 在Flink run命令中增加“-yt”选项来传输keystore和truststore文件到各个执行节点。例如“./bin/flink run **-yt** ssl/ -ys 3 -yn 3 -m yarn-cluster -c com.huawei.SocketWindowWordCount ../lib/flink-eg-1.0.jar --hostname r3-d3 --port 9000” 3. 配置keystore或truststore文件路径为绝对路径时，需要在Flink Client以及Yarn各个节点的该绝对路径上放置keystore或truststore文件。 执行命令中不需要使用“-t”或“-yt”来传输keystore和truststore文件。 4. 将客户端安装节点的业务ip和manager界面浮动ip追加到jobmanager.web.allow-access-address配置中用“,”隔开 ## 三、样例背景 假定用户有某个网站周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Flink的DataStream应用程序实现如下功能： 实时统计总计网购时间超过2个小时的女性网民信息。周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为“,”。 log1.txt：周六网民停留日志。该日志文件在该样例程序中的data目录下获取。 LiuYang,female,20 YuanJing,male,10 GuoYijun,male,5 CaiXuyu,female,50 Liyuan,male,20 FangBo,female,50 LiuYang,female,20 YuanJing,male,10 GuoYijun,male,50 CaiXuyu,female,50 FangBo,female,60 log2.txt：周日网民停留日志。该日志文件在该样例程序中的data目录下获取。 LiuYang,female,20 YuanJing,male,10 CaiXuyu,female,50 FangBo,female,50 GuoYijun,male,5 CaiXuyu,female,50 Liyuan,male,20 CaiXuyu,female,50 FangBo,female,50 LiuYang,female,20 YuanJing,male,10 FangBo,female,50 GuoYijun,male,50 CaiXuyu,female,50 FangBo,female,60 ## 四、样例调试 前提：Linux环境有安装集群客户端，环境准备，参考第一课 数据规划: Flink样例工程的数据存储在Kafka组件中。向Kafka组件发送数据（需要有Kafka权限用户），并从Kafka组件接收数据。确保集群安装完成，包括HDFS、Yarn、Flink和Kafka。 1. 创建Topic。 将Kafka的Broker配置参数“allow.everyone.if.no.acl.found”的值修改为“true” 用户使用Linux命令行创建topic，执行命令前需要使用kinit命令进行人机认证，如：kinit flinkuser。 在kafka客户端目录执行如下指令，ZooKeeper集群信息以自己的为准： ``` bin/kafka-topics.sh --create --zookeeper 10.244.231.91:24002,10.244.230.245:24002,10.244.230.229:24002/kafka --partitions 5 --replication-factor 1 --topic topic3 ``` 2. Kerberos认证配置 修改flink客户端flink-conf.yaml如下配置： security.kerberos.login.keytab: /opt/testclient/user.keytab security.kerberos.login.principal: flinkuser security.kerberos.login.contexts: Client,KafkaClient security.kerberos.login.use-ticket-cache: false 调试步骤： 1. 比对“本地时间和Linux机器时间”与集群时间误都不能超过5分钟 2. 检查linux环境的JDK版本为1.8 3. 配置linux环境的/etc/hosts文件 4. 检查 C:\Windows\System32\drivers\etc\hosts文件中是否包含所有集群节点的域名IP映射信息 5. 在IDEA打开样例代码的FlinkKafkaJavaExample目录，检查SDK配置 6. 默认自动加载依赖，如未加载，则打开后选中pom.xml文件，右键点击“Add As Maven Project”后等待项目自动将依赖下载完毕 7. 在IDEA主页面，选择“File > Project Structures...”进入“Project Structure”页面。 8. 在“Project Structure”页面，选择“Artifacts”，单击“+”并选择“JAR > Empty”。 9. 根据实际情况设置Jar包的名称、类型以及输出路径。 10. 选择“Put into Output Root”。然后单击“Apply”。 11. 在IDEA主页面，选择“Build > Build Artifacts...”。在弹出的菜单中构建我们的jar包 12. 从IDEA项目out目录下的获取到Jar包,拷贝到Flink客户端目录，如“/opt/testclient/Flink/flink”。 13. 将如下包拷到"/opt/testclient/Flink/flink/lib"目录下，若已存在则忽略。 - kafka-clients-2.4.0-hw-ei-302002.jar - flink-connector-kafka_2.11-1.10.0-hw-ei-302002.jar - flink-connector-kafka-base_2.11-1.10.0-hw-ei-302002.jar - flink-dist_2.11-1.10.0-hw-ei-302002.jar 14. 在Linux环境中运行Flink应用程序，需要先启动Flink集群。在Flink客户端下执行yarn session命令，启动flink集群。 例如我们使用ssl相对路径指令为“bin/yarn-session.sh **-t** ssl/ -jm 1024 -tm 1024” 15. 在终端另开一个窗口，进入Flink客户端目录，调用bin/flink run脚本运行代码，例如： a. 生产数据的执行命令启动程序： ``` bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/testclient/Flink/flink/FlinkKafkaJavaExample.jar --topic topic3 -bootstrap.servers 10.244.230.213:21007,10.244.231.48:21007,10.244.231.130:21007 --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka --kerberos.domain.name hadoop.hadoop_arm_802.com ``` b. 消费数据 ``` bin/flink run --class com.huawei.bigdata.flink.examples.ReadFromKafka /opt/testclient/Flink/flink/FlinkKafkaJavaExample.jar --topic topic3 -bootstrap.servers 10.244.230.213:21007,10.244.231.48:21007,10.244.231.130:21007 --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka --kerberos.domain.name hadoop.hadoop_arm_802.com ``` 16. 使用Flink Web页面查看Flink应用程序运行情况。 ## 五、问题互动渠道 FusonInsight 论坛入口 https://bbs.huaweicloud.com/forum/forum-1103-1.html