假定用户开发一个应用程序，用于管理企业中的使用A业务的用户信息，使用Impala客户端实现A业务操作流程如下：

普通表的操作：

1.创建用户信息表user_info。

2.在用户信息中新增用户的学历、职称信息。

3.根据用户编号查询用户姓名和地址。

4.A业务结束后，删除用户信息表

在MRS集群详情页面，点击“前往Manager”

点击“确定”

首次使用MRS Manager会显示如下页面，先后点击“高级->添加例外->确认安全例外”即可

输入账号admin及集群创建时所对应的密码（例如HWcloud@user0）后登陆

点击左侧大数据服务列表中的Impala，进入Impala详情页面

点击“更多->下载客户端”

点击“x86_64->仅保存到如下路径->确定”，Impala客户端将会下载到集群主节点（即公网IP所对应服务器）的/tmp/FusionInsight-Client目录。下载时长约1分钟

打开实验桌面的“Xfce终端”，使用ssh命令连接集群。弹性公网IP见集群详情页。连接成功后可显示“root@node-master...”信息

ssh root@xxx.xxx.xxx.xxx

进入Impala客户端下载目录并解压

cd /tmp/FusionInsight-Client/
tar -vxf FusionInsight_Cluster_1_Impala_Client.tar
tar -vxf FusionInsight_Cluster_1_Impala_ClientConfig.tar

安装Impala客户端至/opt/client目录

cd /tmp/FusionInsight-Client/FusionInsight_Cluster_1_Impala_ClientConfig
./install.sh /opt/client

安装成功后将显示如下信息

切换到Impala安装目录并启动

cd /opt/client
source bigdata_env
impala-shell

启动成功将会显示如下界面

退出Impala环境

quit;

创建用户信息表user_info

vi user_info

按键盘“i”键进入编辑模式，输入用户信息如下

12005000201 A 男 19 A城市
12005000202 B 女 23 B城市
12005000203 C 男 26 C城市
12005000204 D 男 18 D城市
12005000205 E 女 21 E城市
12005000206 F 男 32 F城市
12005000207 G 女 29 G城市
12005000208 H 女 30 H城市
12005000209 I 男 26 I城市
12005000210 J 女 25 J城市

按ECS键退出编辑模式。输入“：wq” 然后回车保存并退出user_info

:wq

运行Impala客户端命令

impala-shell

创建用户信息表user_info并添加相关数据

create table user_info(id string,name string,gender string,age int,addr string);

insert into table user_info(id,name,gender,age,addr) values ("12005000201","A","男",19,"A城市");

在用户信息表user_info中新增用户的学历、职称信息

alter table user_info add columns(education string,technical string);

根据用户编号查询用户姓名和地址

select name,addr from user_info where id='12005000201';

删除用户信息表

drop table user_info;

创建外部表

create external table user_info(id string,name string,gender string,age int,addr string) partitioned by(year string) row format delimited fields terminated by ' ' lines terminated by '\n' stored as textfile location '/hive/user_info';

使用insert语句插入数据

insert into user_info partition(year="2018") values ("12005000201","A","男",19,"A城市");

执行以下SQL语句，查看数据插入成功

select * from user_info;

退出Impala环境

quit;

上传文件至hdfs

hdfs dfs -put user_info /tmp

进入Impala环境

impala-shell

加载数据到表中

load data inpath '/tmp/user_info' into table user_info partition (year='2018');

查询导入数据

select * from user_info;

删除用户信息表

drop table user_info;

假定某个Flink业务每秒就会收到1个消息记录。基于某些业务要求，开发的Flink应用程序实现功能：实时输出带有前缀的消息内容

运行下列代码下载jdk1.8压缩文件

wget https://hcip-materials.obs.cn-north-4.myhuaweicloud.com/jdk-8u341-linux-x64.tar.gz

下载完成后，运行下列命令进行解压：

tar -zxvf jdk-8u341-linux-x64.tar.gz

解压成功后即可，无需修改启动配置文件

打开实验桌面的eclipse，点击“File->New->Project”

选择“Maven Project”，点击“Next”

勾选“Create a simple project…”，点击“Next”

输入：

Group Id: com.huawei

Artifact Id: FlinkExample

点击“Finish”

右上角选择Window标签，在下拉菜单最后一栏中找到Preferences

点击Preferences，在新窗口的左侧找到Java->Compiler

右侧找到Compiler Compliance level 16，点击16的下拉菜单，在列表中找到1.8，选择1.8即可

确认改为1.8之后，点击Apply and Close

然后在下方还能找到Installed JREs标签，点击Add按钮进行配置

在弹出的新窗口处选择Standard VM，然后点击Next

在JRE home处填入/home/user/jdk1.8.0_341，即可自动匹配关联下方信息

确认无误后点击Finish，然后勾选新增的jdk前的框。点击Apply and Close

回到初始界面，看到项目名称下有一个类似JRE System Library [J2SE-1.5]的标签

右键点击该标签，选择Build Path->Configure Build Path

在新窗口点击Add Library

选择JRE System Library，点击Next

此时应该能看到Workspace default JRE (jdk1.8.0_341)，保证勾选后点击Finish

将之前的J2SE-1.5直接删除。选择该模块，在右边找到Remove

删除之后，点击右下角的Apply And Close即可

打开“FlinkExample->pom.xml”文件

将pom.xml文件内容替换如下

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.huawei</groupId>
  <artifactId>FlinkExample</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  
  <properties>
        <flink.version>1.12.0-hw-ei-310003</flink.version>
        <flink.shaded.zookeeper.version>3.5.6-12.0</flink.shaded.zookeeper.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-shaded-hadoop-2</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-shaded-hadoop-2</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-shaded-hadoop-2</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-shaded-zookeeper-3</artifactId>
            <version>${flink.shaded.zookeeper.version}</version>
        </dependency>
    </dependencies>

    <repositories>
        <repository>
            <id>huaweicloud2</id>
            <name>huaweicloud2</name>
            <url>https://mirrors.huaweicloud.com/repository/maven/</url>
        </repository>
        <repository>
            <id>huaweicloud1</id>
            <name>huaweicloud1</name>
            <url>https://repo.huaweicloud.com/repository/maven/huaweicloudsdk/</url>
        </repository>
    </repositories>

    <build>
        <plugins>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>assembly</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

关闭pom.xml，下载第三方依赖程序包需要一定时间，预计约需5分钟左右

在“src/main/java”处右键“new->package”

创建包FlinkDemo，点击“Finish”

在上步FlinkDemo处右键点击，选择“New->Class”创建Java类WriteIntoKafka

package FlinkDemo;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

public class WriteIntoKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        ParameterTool paraTool = ParameterTool.fromArgs(args);
        DataStream<String> messageStream = env.addSource(new SimpleStringGenerator());
        messageStream.addSink(
                new FlinkKafkaProducer<String>(paraTool.get("topic"), new SimpleStringSchema(), paraTool.getProperties()));
        env.execute();
    }

    public static class SimpleStringGenerator implements SourceFunction<String> {
        private static final long serialVersionUID = 2174904787118597072L;
        boolean running = true;
        long i = 0;

        public void run(SourceContext<String> ctx) throws Exception {
            while (running) {
                ctx.collect("element-" + (i++));
                Thread.sleep(1000);
            }
        }

        public void cancel() {
            running = false;
        }
    }
}

在FlinkExample右键，选择“Run As->Maven install”导出Jar包

成功导出Jar包后，可以在target文件夹下找到FlinkExample-0.0.1-SNAPSHOT.jar

将FlinkExample-0.0.1-SNAPSHOT.jar上传至大数据集群/home/omm目录下。弹性公网IP见集群详情页

scp /home/user/eclipse-workspace/FlinkExample/target/FlinkExample-0.0.1-SNAPSHOT.jar root@xxx.xxx.xxx.xxx:/home/omm

类似于Impala客户端安装，需要进入MRS Manager页面，找到对应Flink服务

点击“更多->下载客户端”

点击“x86_64->仅保存到如下路径”，保存路径修改为“/home/omm”。Flink客户端将会下载到集群主节点（即公网IP所对应服务器）的对应目录。下载时长约1分钟

打开实验桌面的“Xfce终端”，使用ssh命令连接集群。弹性公网IP见集群详情页。连接成功后可显示“root@node-master...”信息

ssh root@xxx.xxx.xxx.xxx

修改文件所属用户、用户组信息、操作权限

cd /home/omm
chown omm:wheel FlinkExample-0.0.1-SNAPSHOT.jar
chmod 777 FlinkExample-0.0.1-SNAPSHOT.jar
chmod 777 FusionInsight_Cluster_1_Flink_Client.tar

解压客户端

tar -vxf FusionInsight_Cluster_1_Flink_Client.tar 
tar -vxf FusionInsight_Cluster_1_Flink_ClientConfig.tar

安装Flink客户端至/home/omm目录

cd FusionInsight_Cluster_1_Flink_ClientConfig
./install.sh /home/omm/FlinkClient

安装成功后将显示如下信息

4.6.5 WriteIntoKafka程序运行

在MRS Manager页面，找到对应Kafka服务

点击“实例”即可看到Broker所对应的业务IP信息

同理，在MRS Manager页面，找到对应Zookeeper服务

点击“实例”即可看到Broker所对应的业务IP信息

进入Kakfa客户端目录

cd /opt/Bigdata/components/FusionInsight_HD_8.1.0.1/Kafka/client/install_files/kafka/bin

为脚本添加可执行权限：

chmod 777 kafka-topics.sh
chmod 777 kafka-run-class.sh
chmod 777 kafka-console-producer.sh
chmod 777 kafka-console-consumer.sh

创建Topic

./kafka-topics.sh --create --zookeeper xxx.xxx.xxx.xxx:2181/kafka --topic FlinkTopic --replication-factor 2 --partitions 2

其中“xxx.xxx.xxx.xxx:2181” 为集群中的Zookeeper任一服务地址，端口保持不变

执行命令

./kafka-topics.sh --list --zookeeper xxx.xxx.xxx.xxx:2181/kafka

新建Xfce终端窗口，登录集群。

ssh root@xxx.xxx.xxx.xxx

进入Kafka目录。

cd /opt/Bigdata/components/FusionInsight_HD_8.1.0.1/Kafka/client/install_files/kafka/bin

执行命令创建消费者，这里IP地址为broker的任一业务IP地址。

./kafka-console-consumer.sh --topic FlinkTopic --bootstrap-server xxx.xxx.xxx.xxx:9092 --from-beginning

新建Xfce终端窗口，登录集群

ssh root@xxx.xxx.xxx.xxx

Flink应用处于安全性考虑，需要由非root用户启动

su omm

切换到Flink安装目录

cd /home/omm/FlinkClient

应用环境变量信息

source bigdata_env

检查“FlinkYarnSessionCli”进程是否启动

jps

注意：若无“FlinkYarnSessionCli”进程，则启动Yarn集群

cd Flink/flink/bin
./yarn-session.sh &

运行Flink应用程序

flink run --class FlinkDemo.WriteIntoKafka /home/omm/FlinkExample-0.0.1-SNAPSHOT.jar --topic FlinkTopic --bootstrap.servers xxx.xxx.xxx.xxx:9092

运行成功后可在Kafka消费者窗口看到数据输出

任务完成