这次彻底搞懂impala与flink应用开发【玩转华为云】-4008云顶国际网站

假定用户开发一个应用程序，用于管理企业中的使用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文件内容替换如下

  4.0.0
  com.huawei
  flinkexample
  0.0.1-snapshot
  
  
        1.12.0-hw-ei-310003
        3.5.6-12.0
    
    
        
            org.apache.flink
            flink-core
            ${flink.version}
            
                
                    org.apache.flink
                    flink-shaded-hadoop-2
                
            
        
        
            org.apache.flink
            flink-java
            ${flink.version}
            
                
                    org.apache.flink
                    flink-shaded-hadoop-2
                
            
        
        
            org.apache.flink
            flink-streaming-java_2.11
            ${flink.version}
            
                
                    org.apache.flink
                    flink-shaded-hadoop-2
                
            
        
        
            org.apache.flink
            flink-connector-kafka_2.11
            ${flink.version}
        
        
            org.apache.flink
            flink-shaded-zookeeper-3
            ${flink.shaded.zookeeper.version}
        
    
    
        
            huaweicloud2
            huaweicloud2
            https://mirrors.huaweicloud.com/repository/maven/
        
        
            huaweicloud1
            huaweicloud1
            https://repo.huaweicloud.com/repository/maven/huaweicloudsdk/
        
    
    
        
            
                maven-assembly-plugin
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            assembly
                        
                    
                
            
        
    

关闭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 messagestream = env.addsource(new simplestringgenerator());
        messagestream.addsink(
                new flinkkafkaproducer(paratool.get("topic"), new simplestringschema(), paratool.getproperties()));
        env.execute();
    }
    public static class simplestringgenerator implements sourcefunction {
        private static final long serialversionuid = 2174904787118597072l;
        boolean running = true;
        long i = 0;
        public void run(sourcecontext 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消费者窗口看到数据输出

任务完成