Flink 使用介绍相关文档目录

前言

本篇我们配置Zeppelin环境，实现Zeppelin可视化提交作业到Flink集群，操作Hudi表。

Zeppelin Impersonation

官网链接：Apache Zeppelin 0.10.1 Documentation: Impersonation

通过Zeppelin Impersonation，用户能够以登录Zeppelin的身份来运行作业。

需要的配置：

Linux中创建了目标用户，且开启了zeppelin服务启动用户到目标用户的ssh免密登录。
Zeppelin配置了认证。例如Shiro的固定用户，或者是LDAP等。
对应的interpreter配置为The interpreter will be instantiated Per User in isolated process。并且勾选User Impersonate复选框。

下面我们以配置Flink interpreter的user impersonation为例，讲解整个配置过程。Zeppelin server的启动用户为root，实际执行作业的用户为paul。

创建目标用户并配置免密

我们需要创建paul用户，并配置root用户到本机paul用户的免密。以root用户执行：

# 创建paul用户
useradd paul
# 修改paul用户的密码
passwd paul
# 生成密钥
ssh-keygen
# 配置root到paul的免密
ssh-copy-id paul@localhost # 执行后输入paul用户密码

配置完毕后执行测试：

ssh paul@localhost

如果成功切换到paul用户并且没有弹出密码输入提示，说明配置成功。

配置目标用户的权限

新建的目标用户默认是没有访问HDFS和提交Yarn作业到队列权限的。在进行下一步之前，需要通过Ranger配置paul用户的HDFS和Yarn队列提交作业权限。

Zeppelin配置认证

执行：

mv $ZEPPELIN_HOME/conf/shiro.ini.template $ZEPPELIN_HOME/conf/shiro.ini

然后编辑shiro.ini，修改[users]部分：

[users]
paul = password, admin

这里配置了Zeppelin的paul用户，密码为password，具有admin角色。

然后执行$ZEPPELIN_HOME/bin/zeppelin-daemon.sh restart命令，重启Zeppelin服务。

重启完之后，可以通过web页面使用paul用户登录。

如果要使用LDAP用户认证，请参考：Zeppelin 集成 LDAP(FreeIPA)。

配置用户名密码需要注意的是，直接在shiro.ini中配置明文密码可能会导致敏感内容泄露。为了解决这个问题Shiro支持配置Hash之后的密码，例如：

$shiro1$SHA-256$500000$xZupSS3XInqQoMHK4NAosQ==$bD3n16AfQD2zQ7FdiP2uPh8GWL4dkILfTXAB/lXEZ7c=

支持这个特性需要在shiro.ini的[main]一节中添加：

passwordMatcher = org.apache.shiro.authc.credential.PasswordMatcher
iniRealm.credentialsMatcher = $passwordMatcher

随之而来的问题是我们如何生成这一串Hash之后的密码。可以使用Shiro提供的DefaultPasswordService。接下来举个例子。

新建Maven项目引入如下依赖：

<dependency>
    <groupId>org.apache.shiro</groupId>
    <artifactId>shiro-core</artifactId>
    <version>1.13.0</version>
</dependency>

然后编写如下代码：

DefaultPasswordService defaultPasswordService = new DefaultPasswordService();
String encryptedPassword = defaultPasswordService.encryptPassword("password");
// 这里就是Shiro可以直接使用的Hash之后的password
System.out.println(encryptedPassword);

// 使用这种方式校验密码是否匹配
boolean matched = defaultPasswordService.passwordsMatch("password", encryptedPassword);
System.out.println(matched);

注意：如果运行多次上面的代码，每次得到的encryptedPassword都是不同的，这是因为每次生成的时候会使用随机的salt。校验密码是否匹配不受随机salt的影响，因为可以从Hash之后的密码字符串中解析出salt。

配置Flink Interpreter

点击Zeppelin web界面右上角菜单，选择Interpreter，然后搜索flink。然后点击右侧的edit按钮，修改interpreter配置如下：

image-20220811163809398.png

两个下拉框分别选择Per User和isolated，然后勾选User Impersonate复选框。

Zeppelin Flink HUDI整合使用

这里我们使用Flink on yarn模式，通过Zeppelin编写SQL提交作业到Yarn集群，以paul用户执行。

配置zeppelin-env.sh

执行mv $ZEPPELIN_HOME/conf/zeppelin-env.sh.template $ZEPPELIN_HOME/conf/zeppelin-env.sh，然后编辑这个文件。修改如下内容：

export JAVA_HOME=/opt/zy/jdk8u342-b07 # Zeppelin对使用的JDK版本有特殊要求，需要专门配置
export HADOOP_CONF_DIR=/usr/hdp/3.0.1.0-187/hadoop/conf # core-site.xml等配置文件所在的目录
export USE_HADOOP=true                          # Whether include hadoop jars into zeppelin server process. (true or false)
export FLINK_CONF_DIR=/opt/zy/flink-1.13.2/conf # 必须配置，否则会遇到问题

注意，使用这种方式必须要配置FLINK_CONF_DIR环境变量。否则会出现如下错误，找不到配置文件目录。

 WARN [2022-08-02 23:25:45,537] ({SchedulerFactory43} NotebookServer.java[onStatusChange]:1984) - Job paragraph_1580998080340_1531975932 is finished, status: ERROR, exception: null, result: %text org.apache.zeppelin.interpreter.InterpreterException: org.apache.zeppelin.interpreter.InterpreterException: Fail to open FlinkInterpreter
        at org.apache.zeppelin.interpreter.LazyOpenInterpreter.open(LazyOpenInterpreter.java:76)
        at org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:844)
        at org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:752)
        at org.apache.zeppelin.scheduler.Job.run(Job.java:172)
        at org.apache.zeppelin.scheduler.AbstractScheduler.runJob(AbstractScheduler.java:132)
        at org.apache.zeppelin.scheduler.FIFOScheduler.lambda$runJobInScheduler$0(FIFOScheduler.java:42)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.zeppelin.interpreter.InterpreterException: Fail to open FlinkInterpreter
        at org.apache.zeppelin.flink.FlinkInterpreter.open(FlinkInterpreter.java:80)
        at org.apache.zeppelin.interpreter.LazyOpenInterpreter.open(LazyOpenInterpreter.java:70)
        ... 8 more
Caused by: org.apache.flink.configuration.IllegalConfigurationException: The given configuration directory name '' (/home/paul) does not describe an existing directory.
        at org.apache.flink.configuration.GlobalConfiguration.loadConfiguration(GlobalConfiguration.java:119)
        at org.apache.flink.configuration.GlobalConfiguration.loadConfiguration(GlobalConfiguration.java:94)
        at org.apache.zeppelin.flink.FlinkScalaInterpreter.initFlinkConfig(FlinkScalaInterpreter.scala:216)
        at org.apache.zeppelin.flink.FlinkScalaInterpreter.open(FlinkScalaInterpreter.scala:122)
        at org.apache.zeppelin.flink.FlinkInterpreter.open(FlinkInterpreter.java:75)

配置修改后需要重启Zeppelin服务。

配置Flink interpreter

按照上面的章节打开Flink inteterpreter配置页面。有以下配置项需要特别注意：

FLINK_HOME: Flink安装目录。
HADOOP_CONF_DIR: Hadoop配置文件所在目录，例如/usr/hdp/3.0.1.0-187/hadoop/conf。
HIVE_CONF_DIR: Hive配置文件所在目录，例如/usr/hdp/3.0.1.0-187/hive/conf。
flink.execution.mode: Flink执行模式。采用yarn模式填写yarn。
flink.execution.jars: Flink用户作业依赖的jar，多个jar需要使用逗号分隔（注意不是冒号）。例如connector-kafka, connector-hive, HUDI依赖，table依赖等。这个配置项很重要，否则执行作业的时候会出现class not found等众多问题。
zeppelin.flink.enableHive: 是否启用Flink的Hive支持。我们需要使用Hive metastore，所以配置为true。
zeppelin.flink.hive.version: 配套使用的Hive版本，例如3.1.0。

特别注意：Flink Interpreter执行的时候如果提示缺少jar包，需要修改Flink interpreter的flink.execution.jars为类似如下配置：

/opt/flink-1.13.2/lib/flink-sql-connector-hive-3.1.2_2.11-1.13.2.jar,/opt/flink-1.13.2/lib/flink-connector-files-1.13.2.jar,/opt/flink-1.13.2/lib/flink-table-blink_2.11-1.13.2.jar,/opt/flink-1.13.2/lib/hadoop-mapreduce-client-core-3.1.1.3.0.1.0-187.jar,/opt/flink-1.13.2/lib/hudi-flink1.13-bundle_2.11-0.11.1.jar

也就是说除了Flink自身的jar外，作业依赖的额外jar都必须通过flink.execution.jars配置项引入。仅仅将这些jar放置于Flink的lib目录下是不行的。

运行Flink SQL查询Hudi表数据

在Zeppelin中创建一个新的note，default interperter选择Flink。然后编写如下SQL：

%flink.ssql

show catalogs;
use catalog hive;
show tables;

如果能查到Hive catalog中的表，说明Flink interperter的Hive配置是正确的。

然后我们尝试查询一个已存在的Hudi表，例如hudi_cow。编写如下SQL：

%flink.ssql

use catalog hive;
select * from hudi_cow;

如果上述配置过程正确，这里可以查到数据。

image-20220811170122322.png

同样，create和insert等语句也是可以执行成功的，这里就不再贴出截图。

本博客为作者原创，欢迎大家参与讨论和批评指正。如需转载请注明出处。

Flink Zeppelin Hudi Hive 整合环境配置和使用