华为云用户手册

代码样例 以下代码片段是登录，创建Connection并创建表的示例，在com.huawei.bigdata.hbase.examples包的“HBaseSample”类的HBaseSample方法中。 private TableName tableName = null; private Connection conn = null; public HBaseSample(Configuration conf) throws IOException { this.tableName = TableName.valueOf("hbase_sample_table"); this.conn = ConnectionFactory.createConnection(conf);}

功能分解 根据上述的业务场景进行功能分解，需要开发的功能点如表1所示。 表1 在HBase中开发的功能 序号 步骤 代码实现 1 根据表1中的信息创建表。 请参见创建表。 2 导入用户数据。 请参见插入数据。 3 增加“教育信息”列族，在用户信息中新增用户的学历、职称等信息。 请参见修改表。 4 根据用户编号查询用户姓名和地址。 请参见使用Get读取数据。 5 根据用户姓名进行查询。 请参见使用过滤器Filter。 6 为提升查询性能，创建二级索引或者删除二级索引。 请参见创建二级索引和基于二级索引的查询。 7 用户销户，删除用户信息表中该用户的数据。 请参见删除数据。 8 A业务结束后，删除用户信息表。 请参见删除表。

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“HBaseSample”类的dropTable方法中 public void dropTable() { LOG.info("Entering dropTable."); Admin admin = null; try { admin = conn.getAdmin(); if (admin.tableExists(tableName)) { // Disable the table before deleting it. admin.disableTable(tableName); // Delete table. admin.deleteTable(tableName);//注[1] } LOG.info("Drop table successfully."); } catch (IOException e) { LOG.error("Drop table failed " ,e); } finally { if (admin != null) { try { // Close the Admin object. admin.close(); } catch (IOException e) { LOG.error("Close admin failed " ,e); } } } LOG.info("Exiting dropTable."); }

代码样例 下面代码片段在com.huawei.bigdata.hbase.examples包的“TestMain”类的init方法中。 private static void init() throws IOException { // Default load from conf directory conf = HBaseConfiguration.create(); //In Windows environment String userdir = TestMain.class.getClassLoader().getResource("conf").getPath() + File.separator;[1] //In Linux environment //String userdir = System.getProperty("user.dir") + File.separator + "conf" + File.separator; conf.addResource(new Path(userdir + "core-site.xml"), false); conf.addResource(new Path(userdir + "hdfs-site.xml"), false); conf.addResource(new Path(userdir + "hbase-site.xml"), false); } [1]userdir获取的是编译后资源路径下conf目录的路径。初始化配置用到的core-site.xml、hdfs-site.xml、hbase-site.xml文件，需要放置到"src/main/resources/conf"的目录下。

功能介绍 HBase通过ConnectionFactory.createConnection(configuration)方法创建Connection对象。传递的参数为上一步创建的Configuration。 Connection封装了底层与各实际服务器的连接以及与ZooKeeper的连接。Connection通过ConnectionFactory类实例化。创建Connection是重量级操作，Connection是线程安全的，因此，多个客户端线程可以共享一个Connection。 典型的用法，一个客户端程序共享一个单独的Connection，每一个线程获取自己的Admin或Table实例，然后调用Admin对象或Table对象提供的操作接口。不建议缓存或者池化Table、Admin。Connection的生命周期由调用者维护，调用者通过调用close()，释放资源。

场景说明 假定用户开发一个应用程序，用于管理企业中的使用A业务的用户信息，如表1所示，A业务操作流程如下： 创建用户信息表。 在用户信息中新增用户的学历、职称等信息。 根据用户编号查询用户姓名和地址。 根据用户姓名进行查询。 查询年龄段在[20-29]之间的用户信息。 数据统计，统计用户信息表的人员数、年龄最大值、年龄最小值、平均年龄。 用户销户，删除用户信息表中该用户的数据。 A业务结束后，删除用户信息表。 表1 用户信息 编号 姓名 性别 年龄 地址 12005000201 张三 男 19 广东省深圳市 12005000202 李婉婷 女 23 河北省石家庄市 12005000203 王明 男 26 浙江省宁波市 12005000204 李刚 男 18 湖北省襄阳市 12005000205 赵恩如 女 21 江西省上饶市 12005000206 陈龙 男 32 湖南省株洲市 12005000207 周微 女 29 河南省南阳市 12005000208 杨艺文 女 30 重庆市开县 12005000209 徐兵 男 26 陕西省渭南市 12005000210 肖凯 男 25 辽宁省大连市

接口类型简介 由于HBase本身是由java语言开发出来的，且java语言具有简洁通用易懂的特性，推荐用户使用java语言进行HBase应用程序开发。 HBase采用的接口与Apache HBase保持一致。 HBase通过接口调用，可提供的功能如表1所示。 表1 HBase接口提供的功能 功能 说明 CRUD数据读写功能 增查改删 高级特性 过滤器、二级索引，协处理器 管理功能 表管理、集群管理

HBase简介 HBase是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统。HBase设计目标是用来解决关系型数据库在处理海量数据时的局限性。 HBase使用场景有如下几个特点： 处理海量数据（TB或PB级别以上）。 具有高吞吐量。 在海量数据中实现高效的随机读取。 具有很好的伸缩能力。 能够同时处理结构化和非结构化的数据。 不需要完全拥有传统关系型数据库所具备的ACID特性。ACID特性指原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation，又称独立性）、持久性（Durability）。 HBase中的表具有如下特点： 大：一个表可以有上亿行，上百万列。 面向列：面向列（族）的存储和权限控制，列（族）独立检索。 稀疏：对于为空（null）的列，并不占用存储空间，因此，表可以设计的非常稀疏。

开发流程 本文档主要基于Java API对HBase进行应用开发。 开发流程中各阶段的说明如图1和表1所示。 图1 HBase应用程序开发流程 表1 HBase应用开发的流程说明 阶段 说明 参考文档 准备开发环境 在进行应用开发前，需首先准备开发环境，推荐使用Java语言进行开发，使用IntelliJ IDEA工具，同时完成JDK、Maven等初始配置。 准备开发环境 准备连接集群配置文件 应用程序开发或运行过程中，需通过集群相关配置文件信息连接MRS集群，配置文件通常包括集群组件信息文件，可从已创建好的MRS集群中获取相关内容。 用于程序调测或运行的节点，需要与MRS集群内节点网络互通，同时配置hosts域名信息。 准备连接集群配置文件 配置并导入样例工程 HBase提供了不同场景下的多种样例程序，用户可获取样例工程并导入本地开发环境中进行程序学习。 配置并导入样例工程 根据业务场景开发程序 根据实际业务场景开发程序，调用组件接口实现对应功能。 开发程序 编译并运行程序 开发好的程序编译运行，用户可在本地Windows开发环境中进行程序调测运行，也可以将程序编译为Jar包后，提交到Linux节点上运行。 调测程序 父主题： 概述

常用概念 过滤器 过滤器提供了非常强大的特性来帮助用户提高HBase处理表中数据的效率。用户不仅可以使用HBase中预定义好的过滤器，而且可以实现自定义的过滤器。 协处理器 允许用户执行region级的操作，并且可以使用与RDBMS中触发器类似的功能。 Client 客户端直接面向用户，可通过Java API、HBase Shell或者Web UI访问服务端，对HBase的表进行读写操作。本文中的HBase客户端特指HBase client的安装包，可参考对外接口。 父主题： 概述

问题 Flink内核升级到1.3.0之后，当Kafka调用带有非static的KafkaPartitioner类对象为参数的FlinkKafkaProducer010去构造函数时，运行时会报错。 报错内容如下： org.apache.flink.api.common.InvalidProgramException: The implementation of the FlinkKafkaPartitioner is not serializable. The object probably contains or references non serializable fields.

问题 Flink任务配置State Backend为RocksDB时，运行报如下错误： Caused by: java.lang.UnsatisfiedLinkError: /srv/BigData/hadoop/data1/nm/usercache/***/appcache/application_****/rocksdb-lib-****/librocksdbjni-linux64.so: /lib64/libpthread.so.0: version `GLIBC_2.12` not found (required by /srv/BigData/hadoop/***/librocksdbjni-linux64.so)at java.lang.ClassLoader$NativeLibrary.load(Native Method) at java.lang.ClassLoader.loadLibrary0(ClassLoader.java:1965) at java.lang.ClassLoader.loadLibrary(ClassLoader.java:1890) at java.lang.Runtime.load0(Runtime.java:795) at java.lang.System.load(System.java:1062) at org.rocksdb.NativeLibraryLoader.loadLibraryFromJar(NativeLibraryLoader.java:78)at org.rocksdb.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:56)at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.ensureRocksDBIsLoaded(RocksDBStateBackend.java:734)... 11 more

Flink常用接口 Flink主要使用到如下这几个类： StreamExecutionEnvironment：是Flink流处理的基础，提供了程序的执行环境。 DataStream：Flink用特别的类DataStream来表示程序中的流式数据。用户可以认为它们是含有重复数据的不可修改的集合(collection)，DataStream中元素的数量是无限的。 KeyedStream：DataStream通过keyBy分组操作生成流，数据经过对设置的key值进行分组。 WindowedStream：KeyedStream通过window窗口函数生成的流，设置窗口类型并且定义窗口触发条件，然后在窗口数据上进行一些操作。 AllWindowedStream：DataStream通过window窗口函数生成的流，设置窗口类型并且定义窗口触发条件，然后在窗口数据上进行一些操作。 ConnectedStreams：将两条DataStream流连接起来并且保持原有流数据的类型，然后进行map或者flatMap操作。 JoinedStreams：在窗口上对数据进行等值join操作，join操作是coGroup操作的一种特殊场景。 CoGroupedStreams：在窗口上对数据进行coGroup操作，可以实现流的各种join类型。 图1 Flink Stream的各种流类型转换

提供设置eventtime属性的能力 表6 提供设置eventtime属性的能力的相关接口 API 说明 def assignTimestampsAndWatermarks(assigner: AssignerWithPeriodicWatermarks[T]): DataStream[T] 为了能让event time窗口可以正常触发窗口计算操作，需要从记录中提取时间戳。 def assignTimestampsAndWatermarks(assigner: AssignerWithPunctuatedWatermarks[T]): DataStream[T]

提供Join能力 表12 提供Join能力的相关接口 API 说明 def join[T2](otherStream: DataStream[T2]): JoinedStreams[T, T2] 通过给定的key在一个窗口范围内join两条数据流。 join操作的key值通过where和eaualTo方法进行指定，代表两条流过滤出包含等值条件的数据。 def coGroup[T2](otherStream: DataStream[T2]): CoGroupedStreams[T, T2] 通过给定的key在一个窗口范围内co-group两条数据流。 coGroup操作的key值通过where和eaualTo方法进行指定，代表两条流通过该等值条件进行分区处理。

提供分流能力 表8 提供分流能力的相关接口 API 说明 def split(selector: OutputSelector[T]): SplitStream[T] 传入OutputSelector，重写select方法确定分流的依据(即打标记)，构建SplitStream流。即对每个元素做一个字符串的标记，作为选择的依据，打好标记之后就可以通过标记选出并新建某个标记的流。 def select(outputNames: String*): DataStream[T] 从一个SplitStream中选出一个或多个流。 outputNames指的是使用split方法对每个元素做的字符串标记的序列。

Flink常用接口 Flink主要使用到如下这几个类： StreamExecutionEnvironment：是Flink流处理的基础，提供了程序的执行环境。 DataStream：Flink用类DataStream来表示程序中的流式数据。用户可以认为它们是含有重复数据的不可修改的集合(collection)，DataStream中元素的数量是无限的。 KeyedStream：DataStream通过keyBy分组操作生成流，通过设置的key值对数据进行分组。 WindowedStream：KeyedStream通过window窗口函数生成的流，设置窗口类型并且定义窗口触发条件，然后在窗口数据上进行一些操作。 AllWindowedStream：DataStream通过window窗口函数生成的流，设置窗口类型并且定义窗口触发条件，然后在窗口数据上进行一些操作。 ConnectedStreams：将两条DataStream流连接起来并且保持原有流数据的类型，然后进行map或者flatMap操作。 JoinedStreams：在窗口上对数据进行等值join操作（等值就是判断两个值相同的join，比如a.id = b.id），join操作是coGroup操作的一种特殊场景。 CoGroupedStreams：在窗口上对数据进行coGroup操作，可以实现流的各种join类型。 图1 Flink Stream的各种流类型转换

回答 首先查看ZooKeeper中/flink_base的目录权限是否为：'world,'anyone: cdrwa；如果不是，请修改/flink_base的目录权限为：'world,'anyone: cdrwa，然后继续根据步骤二排查；如果是，请根据步骤二排查。 由于在Flink配置文件中“high-availability.zookeeper.client.acl”默认为“creator”，即谁创建谁有权限，由于原有用户已经使用ZooKeeper上的/flink_base/flink目录，导致新创建的用户访问不了ZooKeeper上的/flink_base/flink目录。 新用户可以通过以下操作来解决问题。 查看客户端的配置文件“conf/flink-conf.yaml”。 修改配置项“high-availability.zookeeper.path.root”对应的ZooKeeper目录，例如：/flink2。 重新提交任务。

回答 由于浏览器所在的计算机IP地址未加到Web访问白名单导致。用户可以通过修改客户端的配置文件“conf/flink-conf.yaml”来解决问题。 确认配置项“jobmanager.web.ssl.enabled”的值是否是“false”，若不是，请修改为“false”。 确认配置项“jobmanager.web.access-control-allow-origin”和“jobmanager.web.allow-access-address”中是否已经添加浏览器所在的计算机IP地址。如果没有添加，可以通过这两项配置项进行添加。例如： jobmanager.web.access-control-allow-origin: 浏览器所在的计算机IP地址jobmanager.web.allow-access-address: 浏览器所在的计算机IP地址

Savepoints相关问题解决方案 用户必须为job中的所有算子均分配ID吗？ 严格的说，用户只给有状态的算子分配IDs即可，因为在savepoint中仅包括有状态的算子的状态，没有状态的算子并不包含在savepoint中。 在实际应用中，强烈建议用户给所有的算子均分配ID，因为有些Flink的内置算子，如window算子是有状态的。具体哪个算子是有状态的，哪个算子是无状态的，不是十分明显。如果用户十分确定某个算子是无状态的，该算子可以不调用uid()方法分配ID。 如果用户在升级作业时新添加一个有状态的算子有什么影响？ 当用户在作业中新添加一个有状态的算子时，由于该算子是新添加的，无保存的旧状态，因此无状态恢复，从0开始运行。 如果用户在升级作业时从作业中删除一个有状态的算子有什么影响？ 默认情况下，savepoint会尝试将所有保存的状态恢复。如果用户使用的savepoint中包含已经删除算子的状态，恢复将会失败。 用户可以通过--allowNonRestoredState(简写为-n)参数跳过恢复已经删除的算子的状态： $ bin/flink run -s savepointPath -n [runArgs] 如果用户重新编排有状态的算子的顺序有什么影响？ 如果用户已经给这些算子分配IDs，那么这些状态会正常恢复。 如果用户没有给这些算子分配IDs， 这些算子将会按新的顺序自动分配新的ID，这将导致状态恢复失败。 如果用户在作业中删除或添加或更改无状态算子的顺序有什么影响？ 如果用户已经给有状态的算子分配ID，那么无状态的算子并不会影响从savepoint进行状态恢复。 如果用户没有分配IDs，有状态算子的IDs由于顺序变化可能会被分配新的IDs，这将导致状态恢复失败。 如果用户在状态恢复时改变了算子的并发度会有什么影响？ 如果Flink版本高于1.2.0且不使用已经废弃的状态API，如checkpointed，用户可以从savepoint中进行状态恢复。否则，无法恢复。 父主题： FAQ

注意事项 如果yarn-session.sh使用-z配置特定的zookeeper的namespace，则在使用flink run时必须使用-yid指出applicationID，使用-yz指出zookeeper的namespace，前后namespace保持一致。 举例： bin/yarn-session.sh -z YARN101 bin/flink run -yid application_****_**** -yz YARN101 examples/streaming/WindowJoin.jar

回答 建议将"blob.storage.directory"配置选项设置成“/tmp”或者“/opt/huawei/Bigdata/tmp”。 当用户将"blob.storage.directory"配置选项设置成自定义目录时，需要手动赋予用户该目录的owner权限。以下以FusionInsight的admin用户为例。 修改Flink客户端配置文件conf/flink-conf.yaml，配置blob.storage.directory: /home/testdir/testdirdir/xxx。 创建目录/home/testdir（创建一层目录即可），设置该目录为admin用户所属。 /home/testdir/下的testdirdir/xxx目录在启动Flink集群时会在每个节点下自动创建。 进入客户端路径，执行命令./bin/yarn-session.sh -jm 2048 -tm 3072，可以看到yarn-session正常启动并且成功创建目录。

数据规划 Flink样例工程的数据存储在Kafka组件中。向Kafka组件发送数据（需要有Kafka权限用户），并从Kafka组件接收数据。 确保集群安装完成，包括HDFS、Yarn、Flink和Kafka。 创建Topic。 创建topic的命令格式： bin/kafka-topics.sh --create --zookeeper {zkQuorum}/kafka --partitions {partitionNum} --replication-factor {replicationNum} --topic {Topic} 表1 参数说明 参数名 说明 {zkQuorum} ZooKeeper集群信息，格式为IP:port。 {partitionNum} topic的分区数。 {replicationNum} topic中每个partition数据的副本数。 {Topic} topic名称。 示例：在Kafka的客户端路径下执行命令，此处以ZooKeeper集群的IP:port是10.96.101.32:2181,10.96.101.251:2181,10.96.101.177:2181,10.91.8.160:2181，Topic名称为topic1的数据为例。 bin/kafka-topics.sh --create --zookeeper 10.96.101.32:2181,10.96.101.251:2181,10.96.101.177:2181,10.91.8.160:2181/kafka --partitions 5 --replication-factor 1 --topic topic1

数据规划 发布者Job使用自定义算子每秒钟产生10000条数据 数据包含两个属性：分别是Int和String类型 配置文件 nettyconnector.registerserver.topic.storage：设置NettySink的IP、端口及并发度信息在第三方注册服务器上的路径（必填），例如： nettyconnector.registerserver.topic.storage: /flink/nettyconnector nettyconnector.sinkserver.port.range：设置NettySink的端口范围（必填），例如： nettyconnector.sinkserver.port.range: 28444-28943 nettyconnector.sinkserver.subnet：设置网络所属域，例如： nettyconnector.sinkserver.subnet: 10.162.0.0/16 接口说明 注册服务器接口 注册服务器用来保存NettySink的IP、端口以及并发度信息，以便NettySource连接使用。为用户提供以下接口： public interface RegisterServerHandler { /** * 启动注册服务器 * @param configuration Flink的Configuration类型 */void start(Configuration configuration) throws Exception;/** *注册服务器上创建Topic节点（目录） * @param topic topic节点名称 */void createTopicNode(String topic) throw Exception;/***将信息注册到某个topic节点（目录）下* @param topic 需要注册到的目录* @param registerRecord 需要注册的信息*/void register(String topic, RegisterRecord registerRecord) throws Exception;/** *删除topic节点 * @param topic 待删除topic */ void deleteTopicNode(String topic) throws Exception;/** *注销注册信息 *@param topic 注册信息所在的topic *@param recordId 待注销注册信息ID */void unregister(String topic, int recordId) throws Exception;/** * 查寻信息* @param 查询信息所在的topic*@recordId 查询信息的ID*/RegisterRecord query(String topic, int recordId) throws Exception;/** * 查询某个Topic是否存在 * @param topic */Boolean isExist(String topic) throws Exception;/** *关闭注册服务器句柄 */void shutdown() throws Exception; 工程基于以上接口提供了ZookeeperRegisterHandler供用户使用。 NettySink算子 Class NettySink(String name, String topic, RegisterServerHandler registerServerHandler,int numberOfSubscribedJobs) name：为本NettySink的名称。 topic：为本NettySink产生数据的Topic，每个不同的NettySink（并发度除外）必须使用不同的TOPIC，否则会引起订阅混乱，数据无法正常分发。 registerServerHandler：为注册服务器的句柄。 numberOfSubscribedJobs：为订阅本NettySink的作业数量，该数量必须是明确的，只有当所有订阅者都连接上NettySink，NettySink才发送数据。 NettySource算子 Class NettySource(String name,String topic,RegisterServerHandler registerServerHandler) name：为本NettySource的名称，该NettySource必须是唯一的（并发度除外），否则，连接NettySink时会出现冲突，导致无法连接。 topic：订阅的NettySink的topic。 registerServerHandler：为注册服务器的句柄。 NettySource的并发度必须与NettySource的并发度相同，否则无法正常创建连接。

场景说明 假定用户有某个网站周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Flink的DataStream应用程序实现如下功能： DataStream应用程序可以在Windows环境和Linux环境中运行。 实时统计总计网购时间超过2个小时的女性网民信息。 周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为“,”。 log1.txt：周六网民停留日志。该日志文件在该样例程序中的data目录下获取。 LiuYang,female,20YuanJing,male,10GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20FangBo,female,50LiuYang,female,20YuanJing,male,10GuoYijun,male,50CaiXuyu,female,50FangBo,female,60 log2.txt：周日网民停留日志。该日志文件在该样例程序中的data目录下获取。 LiuYang,female,20YuanJing,male,10CaiXuyu,female,50FangBo,female,50GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20CaiXuyu,female,50FangBo,female,50LiuYang,female,20YuanJing,male,10FangBo,female,50GuoYijun,male,50CaiXuyu,female,50FangBo,female,60

数据规划 DataStream样例工程的数据存储在文本中。 将log1.txt和log2.txt放置在指定路径下，例如"/opt/log1.txt"和"/opt/log2.txt"。 数据文件若存放在本地文件系统，需在所有部署Yarn NodeManager的节点指定目录放置，并设置运行用户访问权限。 或将数据文件放置于HDFS，并指定程序中读取文件路径HDFS路径，例如"hdfs://hacluster/path/to/file"。

Flink应用程序开发流程 Flink开发流程参考如下步骤： 图1 Flink应用程序开发流程 表1 Flink应用开发的流程说明 阶段 说明 参考文档 了解基本概念 在开始开发应用前，需要了解Flink的基本概念。 常用概念 准备开发和运行环境 Flink的应用程序支持使用Scala、Java两种语言进行开发。推荐使用IDEA工具，请根据指导完成不同语言的开发环境配置。Flink的运行环境即Flink客户端，请根据指导完成客户端的安装和配置。 准备开发和运行环境 准备工程 Flink提供了样例程序，您可以导入样例工程进行程序学习。或者您可以根据指导，新建一个Flink工程。 配置并导入样例工程 新建工程（可选） 根据场景开发工程 提供了Scala、Java两种不同语言的样例工程，帮助用户快速了解Flink各部件的编程接口。 开发程序 编译并运行程序 指导用户将开发好的程序编译并提交运行。 编包并运行程序 查看程序运行结果 程序运行结果会写在用户指定的路径下，用户还可以通过UI查看应用运行情况。 查看调测结果 调优程序 您可以根据程序运行情况，对程序进行调优，使其性能满足业务场景需求。 调优完成后，请重新进行编译和运行。 组件操作指南中的“Flink性能调优”

操作步骤 打开IDEA工具，选择“Create New Project”。 图1 创建工程 在“New Project”页面，选择“Scala”开发环境，并选择“Scala Module”，然后单击“Next”。 如果您需要新建Java语言的工程，选择对应参数即可。 图2 选择开发环境 在工程信息页面，填写工程名称和存放路径，设置JDK版本和Scala SDK，然后单击“Finish”完成工程创建。 图3 填写工程信息

常用接口 YARN常用的Java类有如下几个。 ApplicationClientProtocol 用于Client与ResourceManager之间。Client通过该协议可实现将应用程序提交到ResourceManager上，查询应用程序的运行状态或者中止应用程序等功能。 表1 ApplicationClientProtocol常用方法 方法 说明 forceKillApplication(KillApplicationRequest request) Client通过此接口请求RM中止一个已提交的任务。 getApplicationAttemptReport(GetApplicationAttemptReportRequest request) Client通过此接口从RM获取指定ApplicationAttempt的报告信息。 getApplicationAttempts(GetApplicationAttemptsRequest request) Client通过此接口从RM获取所有ApplicationAttempt的报告信息。 getApplicationReport(GetApplicationReportRequest request) Client通过此接口从RM获取某个应用的报告信息。 getApplications(GetApplicationsRequest request) Client通过此接口从RM获取满足一定过滤条件的应用的报告信息。 getClusterMetrics(GetClusterMetricsRequest request) Client通过此接口从RM获取集群的Metrics。 getClusterNodes(GetClusterNodesRequest request) Client通过此接口从RM获取集群中的所有节点信息。 getContainerReport(GetContainerReportRequest request) Client通过此接口从RM获取某个Container的报告信息。 getContainers(GetContainersRequest request) Client通过此接口从RM获取某个ApplicationAttemp的所有Container的报告信息。 getDelegationToken(GetDelegationTokenRequest request) Client通过此接口获取授权票据，用于container访问相应的service。 getNewApplication(GetNewApplicationRequest request) Client通过此接口获取一个新的应用ID号，用于提交新的应用。 getQueueInfo(GetQueueInfoRequest request) Client通过此接口从RM中获取队列的相关信息。 getQueueUserAcls(GetQueueUserAclsInfoRequest request) Client通过此接口从RM中获取当前用户的队列访问权限信息。 moveApplicationAcrossQueues(MoveApplicationAcrossQueuesRequest request) 移动一个应用到新的队列。 submitApplication(SubmitApplicationRequest request) Client通过此接口提交一个新的应用到RM。 ApplicationMasterProtocol 用于ApplicationMaster与ResourceManager之间。ApplicationMaster使用该协议向ResourceManager注册、申请资源、获取各个任务的运行情况等。 表2 ApplicationMasterProtocol常用方法 方法 说明 allocate(AllocateRequest request) AM通过此接口提交资源分配申请。 finishApplicationMaster(FinishApplicationMasterRequest request) AM通过此接口通知RM其运行成功或者失败。 registerApplicationMaster(RegisterApplicationMasterRequest request) AM通过此接口向RM进行注册。 ContainerManagementProtocol 用于ApplicationMaster与NodeManager之间。ApplicationMaster使用该协议要求NodeManager启动/中止Container或者查询Container的运行状态。 表3 ContainerManagementProtocol常用方法 方法 说明 getContainerStatuses(GetContainerStatusesRequest request) AM通过此接口向NM请求Containers的当前状态信息。 startContainers(StartContainersRequest request) AM通过此接口向NM提供需要启动的containers列表的请求。 stopContainers(StopContainersRequest request) AM通过此接口请求NM停止一系列已分配的Containers。

架构 Flink架构如图2所示。 图2 Flink架构 Flink整个系统包含三个部分： Client Flink Client主要给用户提供向Flink系统提交用户任务（流式作业）的能力。 TaskManager Flink系统的业务执行节点，执行具体的用户任务。TaskManager可以有多个，各个TaskManager都平等。 JobManager Flink系统的管理节点，管理所有的TaskManager，并决策用户任务在哪些Taskmanager执行。JobManager在HA模式下可以有多个，但只有一个主JobManager。 Flink系统提供的关键能力： 低时延 提供ms级时延的处理能力。 Exactly Once 提供异步快照机制，保证所有数据真正只处理一次。 HA JobManager支持主备模式，保证无单点故障。 水平扩展能力 TaskManager支持手动水平扩展。