1 APACHE KUDU简述

Apache Kudu是一个针对 Apache Hadoop 平台而开发的列式存储管理器。Kudu 共享 Hadoop 生态系统应用的常见技术特性: 它在 commodity hardware（商品硬件）上运行，horizontally scalable（水平可扩展），并支持 highly available（高可用）性操作。
官方网站：https://kudu.apache.org

1.1 APACHE KUDU架构组成介绍

1.1.1 基本概念

列数据存储 Columnar Data Store
Kudu是一种列数据储存结构，以强类型的列（strong-type column）储存数据。
高效读取
可选择单个列或者某个列的一部分来访问，可以在满足本身查询需要的状态下，选择最少的磁盘或者存储块来访问，相对于基于行的存储，更节省访问资源，更高效。
数据比较
由于给定的某一个列当中都是同样类型的数据，所以对于同一个量级的数据比较时，这种存储方式比混合类型存储的更具优势。
表Table
同理，一种数据设计模式schema，根据primary key来排序组织。一个表可以被分到若干个分片中，称为tablet。
分片Tablet
一个tablet是指表上一段连续的segment。一个特定的tablet会被复制到多个tablet服务器上，其中一个会被认为是leader tablet。每一个备份tablet都可以支持读取、写入请求。
分片服务器 Tablet Server
负责为客户端储存和提供tablets。只有Leader Tablet可以写入请求，其他的tablets只能执行请求。
Master
Master负责追踪tablets、tablet severs、catalog table和其他与集群相关的metadata。另外也为客户端协调metadata的操作。
Raft Consensus算法
类似半数选举机制。例如，如果3个副本中有2个副本或5个副本中有3个副本可用，则可用
Catalog Table
Kudu的metadata的中心位置，存储表和tablet的信息，客户端可以通过master用客户端api来访问。
逻辑复制 Logical Replication
Kudu并是不是在硬盘数据上做复制的，而是采取了逻辑复制的办法，这有以下一些好处：
• 尽管insert和update需要通过网络对数据做transmit，但是delete操作不需要移动任何数据。Delete操作的请求会发送到每一个tablet server上，在本地做删除操作。
• 普通的物理操作，比如数据压缩，并不需要通过网络做数据transmit，但不同于HDFS，每个请求都需要通过网络把请求传送到各个备份节点上来满足操作需要。
• 每个备份不需要同时进行操作，降低写入压力，避免高延时。
随机写入效率
在内存中每个tablet分区维护一个MemRowSet来管理最新更新的数据，当尺寸大于一定大小之后会flush到磁盘上行成DiskRowSet，多个DiskRowSet会在适当的时候做归并操作。 这些被flush到磁盘的DiskRowSet数据分为两种，一种是Base数据，按列式存储格式存在，一旦生成不再修改，另一种是Delta文件，储存Base中有更新的数据，一个Base文件可以对应多个Delta文件。
Delta文件的存在使得检索过程需要额外的开销，这些Delta文件是根据被更新的行在Base文件中的位移来检索的，而且做合并时也是有选择的进行。
此外DRS（Distributed Resource Scheduler）自身也会合并，为了保障检索延迟的可预测性。Kudu的DRS默认以32MB为单位进行拆分，Compaction过程是为了对内容进行排序重组，减少不同DRS之间key的overlap，进而在检索的时候减少需要参与检索的DRS的数量。

1.1.2 整体框架

1.2 APACHE KUDU的特点

Kudu的主要特点:

1. OLAP工作的快速处理；
2. 与MapReduce，Spark和其他Hadoop生态系统组件集成
3. 与Apache Impala(incubating)紧密集成，使其与Apache Parquet 一起使用 HDFS 成为一个很好的可变的替代方案；
4. 同时运行顺序和随机工作负载的强大性能。
5. 高可用性。tablet server和master利用Raft Consensus算法保证节点的可用。例如，如果3个副本中有2个副本或5个副本中有3个副本可用，则可用
6. 使用 Cloudera Manager 轻松维护和管理(与impala同为CM开发)

1.3 APACHE KUDU的缺陷

Kudu的优点同时也是Kudu的缺点。只有在与CDH、Impala结合使用的时候，它才能发挥最大的优势，单独摘出来在Apache集群中部署使用的时候，优势会降低很多。

1.4 APACHE KUDU应用场景

1. 实时更新的应用。刚刚到达的数据就马上要被终端用户使用访问到。
2. 时间序列相关的应用，需要同时支持：
   根据海量历史数据查询；
   必须非常快地返回关于单个实体的细粒度查询。
3. 实时预测模型的应用，支持根据所有历史数据周期地更新模型。

1.5 APACHE KUDU各个端口

端口名称	默认端口	说明
ntp	123	ntp端口号
Kudu-master	7051	主从机通信他端口号
kudu	8051	Kudu Web端口

1.6 中间件版本选取

中间件名称	版本号
CentOS	CentOS 6.8
Java	1.8.0_121
ntp	4
Kudu	1.2.0

2 APACHE KUDU部署

2.1 环境准备

本次技术实践安装Kudu集群，安装在3台虚拟机上：hadoop102、hadoop103、hadoop104
由于Kudu在Linux(CentOS或者Ubantu)需要编译，这里我直接下载rpm包进行安装部署。不需要kudu-debuginfo包，已隐去。

本次Kudu的安装规划：

host	kudu-master	kudu-tserver
hadoop102	*
hadoop103		*
hadoop104		*

2.1.1 CentOS6.8

CentOS6.8安过程省略。预先创建用户/用户组zhouchen
预先安装jdk1.8.0_92 +
预先安装ntp

2.1.2 关闭防火墙-root

针对CentOS7以下
1.查看防火墙状态
service iptables status
2.停止防火墙
service iptables stop
3.启动防火墙
service iptables start

2.1.3 ntp

Kudu安装需要用到ntp服务
1.检查ntp服务状态

[root@hadoop102 software]# service ntpd status
ntpd (pid  6014) 正在运行...

2.检查ntp时间

[root@hadoop102 software]# ntptime
ntp_gettime() returns code 0 (OK)time e2b93ca0.4836eb34  Wed, Jul 15 2020 16:28:16.282, (.282088806),maximum error 513259 us, estimated error 10946 us, TAI offset 0
ntp_adjtime() returns code 0 (OK)modes 0x0 (),offset 1012.483 us, frequency -5.592 ppm, interval 1 s,maximum error 513259 us, estimated error 10946 us,status 0x2001 (PLL,NANO),time constant 7, precision 0.001 us, tolerance 500 ppm,

2.2 集群安装

2.2.1 rpm安装

1.主机安装(hadoop102)
主机需要的安装包：
kudu-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm
kudu-client0-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm
kudu-client-devel-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm
kudu-master-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm

[root@hadoop102 software]# rpm -ivh --nodeps kudu*
warning: kudu-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm: Header V4 DSA/SHA1 Signature, key ID e8f86acd: NOKEY
Preparing...                ########################################### [100%]1:kudu-client0           ########################################### [ 25%]2:kudu                   ########################################### [ 50%]3:kudu-master            ########################################### [ 75%]4:kudu-client-devel      ########################################### [100%]

2.从机安装(hadoop103/hadoop104)
从机需要的安装包：
kudu-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm
kudu-client0-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm
kudu-client-devel-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm
kudu-tserver-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm

[root@hadoop103 software]# rpm -ivh --nodeps kudu*
warning: kudu-1.2.0+cdh5.10.0+0-1.cdh5.10.0.p0.56.el6.x86_64.rpm: Header V4 DSA/SHA1 Signature, key ID e8f86acd: NOKEY
Preparing...                ########################################### [100%]1:kudu-client0           ########################################### [ 25%]2:kudu                   ########################################### [ 50%]3:kudu-tserver           ########################################### [ 75%]4:kudu-client-devel      ########################################### [100%]

2.2.2 集群配置

*三台机器都需要修改如下配置：
1.修改Kudu主机配置(hadoop102)

[root@hadoop102 software]# vim /etc/kudu/conf/master.gflagfile
# Do not modify these two lines. If you wish to change these variables,
# modify them in /etc/default/kudu-master.
--fromenv=rpc_bind_addresses
--fromenv=log_dir--fs_wal_dir=/opt/module/kudu-1.2.0/master
--fs_data_dirs=/opt/module/kudu-1.2.0/master

2.修改Kudu从机配置(hadoop103/hadoop104)

[root@hadoop103 software]# vim /etc/kudu/conf/tserver.gflagfile
# Do not modify these two lines. If you wish to change these variables,
# modify them in /etc/default/kudu-tserver.
--fromenv=rpc_bind_addresses
--fromenv=log_dir--fs_wal_dir=/opt/module/kudu-1.2.0/tserver
--fs_data_dirs=/opt/module/kudu-1.2.0/tserver
--tserver_master_addrs=hadoop102:7051

3.创建主机目录

[root@hadoop102 module]# mkdir -p /opt/module/kudu-1.2.0/master
[root@hadoop102 module]# chown zhouchen:zhouchen -R kudu-1.2.0/
[root@hadoop102 module]# chmod 777 -R kudu-1.2.0/

4.创建从机目录

[root@hadoop103 module]# mkdir -p /opt/module/kudu-1.2.0/tserver
[root@hadoop103 module]# chown zhouchen:zhouchen -R kudu-1.2.0/
[root@hadoop103 module]# chmod 777 -R kudu-1.2.0/

2.2.3 集群启动

主机启动(hadoop102)：

[root@hadoop102 module]# sudo service kudu-master start
Started Kudu Master Server (kudu-master):                  [确定]

从机启动(hadoop103/hadoop104)：

[root@hadoop102 kudu-1.2.0]# ssh hadoop103 sudo service kudu-tserver start
Started Kudu Tablet Server (kudu-tserver):                 [确定]
[root@hadoop102 kudu-1.2.0]# ssh hadoop104 sudo service kudu-tserver start
Started Kudu Tablet Server (kudu-tserver):                 [确定]

2.2.4 安装检查

1.界面访问 http://Hadoop102:8051/

2.查看服务状态
主机状态：

[zhouchen@hadoop102 ~]$ sudo service kudu-master status
Kudu Master Server is running                              [确定]

从机状态：

[zhouchen@hadoop102 ~]$ ssh hadoop103 sudo service kudu-tserver status
Kudu Tablet Server is running                              [确定][zhouchen@hadoop102 ~]$  ssh hadoop104 sudo service kudu-tserver status
Kudu Tablet Server is running                              [确定]

3 APACHE KUDU的基本操作

1. CREATE/ALTER/DROP TABLE
   Impala支持使用Kudu作为持久层创建、修改和删除表。这些表与Impala中的其他表遵循相同的内部/外部方法，允许灵活的数据摄入和查询
2. INSERT
   可以使用与任何其他Impala表相同的语法（例如使用HDFS或HBase进行持久化的语法）将数据插入Impala中的Kudu表。
3. UPDATE/DELETE
   Impala支持UPDATE和DELETE SQL命令来逐行或批量修改Kudu表中的现有数据。 选择SQL命令的语法，使其与现有标准尽可能兼容。 除了简单的DELETE或UPDATE命令之外，还可以在子查询中使用FROM子句指定复杂的联接。
4. 灵活的分区
   与Hive中的表分区类似，Kudu允许您通过哈希或范围将预动态拆分表动态划分为预定义数量的数位板，以便在整个群集中平均分配写入和查询。 您可以按任意数量的主键列，任意数量的哈希和可选的拆分行列表进行分区。 请参阅架构设计。
5. 平行扫描
   为了在现代硬件上实现最高的性能，Impala使用的Kudu客户端可以并行扫描多个分片。

[zhouchen@hadoop102 lib]$ kudu table list;
Invalid argument: must provide master_addresses

6. 高效查询
   在可能的情况下，Impala将谓词评估下推至Kudu，以便对谓词的评估尽可能接近数据。 在许多工作负载中，查询性能可与Parquet相当

4 APACHE KUDU与IMPALA

通过Impala使用Kudu可以新建内部表和外部表两种：
• 内部表（Internal Table）：事实上是属于Impala管理的表，当删除时会确确实实地删除表结构和数据。在Impala中建表时，默认建的是内部表。
• 外部表（External Table）：不由Impala管理，当删除这个表时，并不能从源位置将其删除，只是接触了Kudu到Impala之间对于这个表的关联关系

4.1 创建一个简单的KUDU表

CREATE TABLE my_second_table
(id BIGINT,name STRING,PRIMARY KEY(id)
)
PARTITION BY HASH PARTITIONS 16
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses' = 'hadoop102:7051', 'kudu.table_name' = 'my_second_table','kudu.num_tablet_replicas' = '1'
);

建表语句中，默认第一个就是Primary Key，是个not null列，在后面的kudu.key_columns中列出，这边至少写一个。
• storage_handler：选择通过Impala访问kudu的机制，必须填成com.cloudera.kudu.hive.KuduStorageHandler
• kudu.table_name：Impala为Kudu建（或者关联的）的表名
• kudu.master_addresses：Impala需要访问的Kudu master列表
• kudu.key_columns：Primary key列表

4.2 插入数据

INSERT INTO my_first_table VALUES (1, "john"), (2, "jane"), (3, "jim");

Impala默认一次同时最多插入1024条记录，作为一个batch

4.3 更新数据

UPDATE my_first_table SET name="bob" where id = 3;

4.4 删除数据

DELETE FROM my_first_table WHERE id < 3;

4.5 修改表属性

ALTER TABLE my_first_table RENAME TO employee;//重命名ALTER TABLE employee SET TBLPROPERTIES('kudu.master_addresses' = 'hadoop102:7051');//更改kudu master addressALTER TABLE employee SET TBLPROPERTIES('EXTERNAL' = 'TRUE');//将内部表变为外部表

4.6 删除表

由于Kudu是作为存储中间件，而impala只是作为操作中间件，所以需要同时在Kudu与impala删除表。

1. Kudu删除表

[zhouchen@hadoop102 lib]$ kudu table delete hadoop102 my_first_table

2. Impala删除表

[hadoop102:21000] > drop table my_first_table;

5 APACHE KUDU与SPARK

5.1 SPARK DF操作KUDU

5.1.1 所需依赖

添加kudu依赖jar包

<dependency><groupId>org.apache.kudu</groupId><artifactId>kudu-client</artifactId><version>1.10.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-spark2 -->
<dependency><groupId>org.apache.kudu</groupId><artifactId>kudu-spark2_2.11</artifactId><version>1.7.0</version>
</dependency>

5.1.2 Spark创建Kudu表

import java.utilimport org.apache.kudu.client.CreateTableOptions
import org.apache.kudu.spark.kudu._
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}//创建kudu表
object KuduSparkTest1 {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]")val spark = SparkSession.builder().config(sparkConf).getOrCreate()val sparkConetxt = spark.sparkContextval kuduContext = new KuduContext("hadoop002,hadoop003", sparkConetxt)val tableFields = (Array(new StructField("id", IntegerType, false), new StructField("name", StringType)))//id为主键val arrayList = new util.ArrayList[String]()arrayList.add("id")val b = new CreateTableOptions().setNumReplicas(1).addHashPartitions(arrayList, 3)kuduContext.createTable("test_table", StructType(tableFields), Seq("id"), b)}
}

将代码打成jar包 使用集群运行 运行完毕后查看UI界面

成功创建test_table测试表

5.1.3 Spark查询Kudu表

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession//查询kudu
object KuduSparkTest2 {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]")val spark = SparkSession.builder().config(sparkConf).getOrCreate()val df = spark.read.options(Map("kudu.master" -> "hadoop001 ", "kudu.table" -> "test_table")).format("org.apache.kudu.spark.kudu").load()df.show()}
}

查询出来返回值是个dataframe，之后就可以使用dataframe api进行操作

5.1.4 Spark增删改Kudu表

object KuduSparkTest3 {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf()val spark = SparkSession.builder().config(sparkConf).getOrCreate()val sparkContext = spark.sparkContextval kuduContext = new KuduContext("hadoop002,hadoop003", sparkContext)val testTableDF = spark.read.options(Map("kudu.master" -> "hadoop001 ", "kudu.table" -> "test_table")).format("org.apache.kudu.spark.kudu").load()val flag = kuduContext.tableExists("test_table") //判断表是否存在if (flag) {import spark.implicits._val tuple = (1, "张三")val df = sparkContext.makeRDD(Seq(tuple)).toDF("id","name")kuduContext.insertRows(df, "test_table") //往test_table表中插入 id为1  name为张三的数据testTableDF.show()val tuple2 = (1, "李四")val df2 = sparkContext.makeRDD(Seq(tuple2)).toDF("id","name")kuduContext.updateRows(df2, "test_table") //将test_table表中主键id为1 的数据name值修改为李四testTableDF.show()kuduContext.deleteRows(df2, "test_table") //将test_table表中的主键id为1 name值为李四的数据删除testTableDF.show()}}
}

效果图

5.2 SPARK STREAMING实时写KUDU例子

注意：spark创建的kudu表在impala里不会显示，但确实存在，在impala创建外部表指定kudu表即可

5.2.1 pom.xml配置

<?xml version="1.0" encoding="UTF-8"?>
<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 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>education-online</artifactId><groupId>com.atguigu</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>com_atguigu_spark_kudu</artifactId><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><scope>provided</scope><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><scope>provided</scope><version>${spark.version}</version></dependency><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><scope>provided</scope><version>${scala.version}</version></dependency><dependency><groupId>org.apache.kudu</groupId><artifactId>kudu-client</artifactId><version>1.10.0</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.29</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-spark2 --><dependency><groupId>org.apache.kudu</groupId><artifactId>kudu-spark2_2.11</artifactId><version>1.7.0</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version>1.1.16</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.11</artifactId><!--            <scope>provided</scope>--><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.11</artifactId><scope>provided</scope><version>${spark.version}</version></dependency></dependencies><build><plugins><plugin><groupId>org.scala-tools</groupId><artifactId>maven-scala-plugin</artifactId><version>2.15.1</version><executions><execution><id>compile-scala</id><goals><goal>add-source</goal><goal>compile</goal></goals></execution><execution><id>test-compile-scala</id><goals><goal>add-source</goal><goal>testCompile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><configuration><archive><manifest></manifest></archive><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration></plugin></plugins></build>
</project>

5.2.2 先使用impala创建kudu表

[hadoop003:21000] > CREATE TABLE register_table(> id INTEGER,> servicetype STRING,> count BIGINT,> PRIMARY KEY(id)> )> PARTITION BY HASH PARTITIONS 3> STORED AS KUDU;

5.2.3 准备工具类

package util;import java.io.InputStream;
import java.util.Properties;/**** 读取配置文件工具类*/
public class ConfigurationManager {private static Properties prop = new Properties();static {try {InputStream inputStream = ConfigurationManager.class.getClassLoader().getResourceAsStream("comerce.properties");prop.load(inputStream);} catch (Exception e) {e.printStackTrace();}}//获取配置项public static String getProperty(String key) {return prop.getProperty(key);}//获取布尔类型的配置项public static boolean getBoolean(String key) {String value = prop.getProperty(key);try {return Boolean.valueOf(value);} catch (Exception e) {e.printStackTrace();}return false;}
}package util;import com.alibaba.druid.pool.DruidDataSourceFactory;import javax.sql.DataSource;
import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;/*** 德鲁伊连接池*/
public class DataSourceUtil implements Serializable {public static DataSource dataSource = null;static {try {Properties props = new Properties();props.setProperty("url", ConfigurationManager.getProperty("jdbc.url"));props.setProperty("username", ConfigurationManager.getProperty("jdbc.user"));props.setProperty("password", ConfigurationManager.getProperty("jdbc.password"));props.setProperty("initialSize", "5"); //初始化大小props.setProperty("maxActive", "10"); //最大连接props.setProperty("minIdle", "5");  //最小连接props.setProperty("maxWait", "60000"); //等待时长props.setProperty("timeBetweenEvictionRunsMillis", "2000");//配置多久进行一次检测,检测需要关闭的连接 单位毫秒props.setProperty("minEvictableIdleTimeMillis", "600000");//配置连接在连接池中最小生存时间 单位毫秒props.setProperty("maxEvictableIdleTimeMillis", "900000"); //配置连接在连接池中最大生存时间 单位毫秒props.setProperty("validationQuery", "select 1");props.setProperty("testWhileIdle", "true");props.setProperty("testOnBorrow", "false");props.setProperty("testOnReturn", "false");props.setProperty("keepAlive", "true");props.setProperty("phyMaxUseCount", "100000");
//            props.setProperty("driverClassName", "com.mysql.jdbc.Driver");dataSource = DruidDataSourceFactory.createDataSource(props);} catch (Exception e) {e.printStackTrace();}}//提供获取连接的方法public static Connection getConnection() throws SQLException {return dataSource.getConnection();}// 提供关闭资源的方法【connection是归还到连接池】// 提供关闭资源的方法 【方法重载】3 dqlpublic static void closeResource(ResultSet resultSet, PreparedStatement preparedStatement,Connection connection) {// 关闭结果集// ctrl+alt+m 将java语句抽取成方法closeResultSet(resultSet);// 关闭语句执行者closePrepareStatement(preparedStatement);// 关闭连接closeConnection(connection);}private static void closeConnection(Connection connection) {if (connection != null) {try {connection.close();} catch (SQLException e) {e.printStackTrace();}}}private static void closePrepareStatement(PreparedStatement preparedStatement) {if (preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {e.printStackTrace();}}}private static void closeResultSet(ResultSet resultSet) {if (resultSet != null) {try {resultSet.close();} catch (SQLException e) {e.printStackTrace();}}}
}package utilimport java.sql.{Connection, PreparedStatement, ResultSet}trait QueryCallback {def process(rs: ResultSet)
}class SqlProxy {private var rs: ResultSet = _private var psmt: PreparedStatement = _/*** 执行修改语句** @param conn* @param sql* @param params* @return*/def executeUpdate(conn: Connection, sql: String, params: Array[Any]): Int = {var rtn = 0try {psmt = conn.prepareStatement(sql)if (params != null && params.length > 0) {for (i <- 0 until params.length) {psmt.setObject(i + 1, params(i))}}rtn = psmt.executeUpdate()} catch {case e: Exception => e.printStackTrace()}rtn}/*** 执行查询语句* 执行查询语句** @param conn* @param sql* @param params* @return*/def executeQuery(conn: Connection, sql: String, params: Array[Any], queryCallback: QueryCallback) = {rs = nulltry {psmt = conn.prepareStatement(sql)if (params != null && params.length > 0) {for (i <- 0 until params.length) {psmt.setObject(i + 1, params(i))}}rs = psmt.executeQuery()queryCallback.process(rs)} catch {case e: Exception => e.printStackTrace()}}def shutdown(conn: Connection): Unit = DataSourceUtil.closeResource(rs, psmt, conn)
}

5.2.4 Spark Streaming写Kudu

import java.sql.ResultSet
import java.{lang, util}import _root_.util.{DataSourceUtil, QueryCallback, SqlProxy}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kudu.client.KuduPredicate.ComparisonOp
import org.apache.kudu.client.{KuduClient, KuduPredicate}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}import scala.collection.mutableobject KuduSparkTest4 {private val groupid = "register_group_test"private val KUDU_MASTERS = "hadoop102"def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName(this.getClass.getSimpleName).set("spark.streaming.kafka.maxRatePerPartition", "100").set("spark.streaming.stopGracefullyOnShutdown", "true").set("spark.streaming.backpressure.enabled", "true")val ssc = new StreamingContext(conf, Seconds(3))val sparkContext = ssc.sparkContextval topics = Array("register_topic")val kafkaMap: Map[String, Object] = Map[String, Object]("bootstrap.servers" -> "hadoop001:9092,hadoop002:9092,hadoop003:9092","key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> groupid,"auto.offset.reset" -> "earliest","enable.auto.commit" -> (false: lang.Boolean))sparkContext.hadoopConfiguration.set("fs.defaultFS", "hdfs://nameservice1") //设置高可用地址sparkContext.hadoopConfiguration.set("dfs.nameservices", "nameservice1") //设置高可用地址val sqlProxy = new SqlProxyval offsetMap = new mutable.HashMap[TopicPartition, Long]()val client = DataSourceUtil.getConnectiontry {sqlProxy.executeQuery(client, "select * from `offset_manager` where groupid=?", Array(groupid), new QueryCallback {override def process(rs: ResultSet): Unit = {while (rs.next()) {val model = new TopicPartition(rs.getString(2), rs.getInt(3))val offset = rs.getLong(4)offsetMap.put(model, offset)}rs.close() //关闭游标}})} catch {case e: Exception => e.printStackTrace()} finally {sqlProxy.shutdown(client)}//设置kafka消费数据的参数  判断本地是否有偏移量  有则根据偏移量继续消费 无则重新消费val stream: InputDStream[ConsumerRecord[String, String]] = if (offsetMap.isEmpty) {KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaMap))} else {KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaMap, offsetMap))}val resultDStream = stream.mapPartitions(partitions => {partitions.map(item => {val line = item.value()val arr = line.split("\t")val id = arr(1)val app_name = id match {case "1" => "PC_1"case "2" => "APP_2"case _ => "Other_3"}(app_name, 1)})}).reduceByKey(_ + _)resultDStream.foreachRDD(rdd => {rdd.foreachPartition(parititon => {val kuduClient = new KuduClient.KuduClientBuilder(KUDU_MASTERS).build //获取kudu连接val kuduTable = kuduClient.openTable("impala::default.register_table") //获取kudu tableval schema = kuduTable.getSchema//根据当前表register_table  将相应需要查询的列放到arraylist中  表中有id servicetype count字段val projectColumns = new util.ArrayList[String]()projectColumns.add("id")projectColumns.add("servicetype")projectColumns.add("count")parititon.foreach(item => {var resultCount: Long = item._2 //声明结果值  默认为当前批次数据val appname = item._1.split("_")(0)val id = item._1.split("_")(1).toIntval eqPred = KuduPredicate.newComparisonPredicate(schema.getColumn("servicetype"),ComparisonOp.EQUAL, appname); //先根据设备名称过滤  过滤条件为等于app_name的数据val kuduScanner = kuduClient.newScannerBuilder(kuduTable).addPredicate(eqPred).build()while (kuduScanner.hasMoreRows) {val results = kuduScanner.nextRows()while (results.hasNext) {val result = results.next()//符合条件的数据有值则 和当前批次数据进行累加val count = result.getLong("count") //获取表中count值resultCount += count}}//最后将结果数据重新刷新到kuduval kuduSession = kuduClient.newSession()val upset = kuduTable.newUpsert() //调用upset 当主键存在数据进行修改操作 不存在则新增val row = upset.getRowrow.addInt("id", id)row.addString("servicetype", appname)row.addLong("count", resultCount)kuduSession.apply(upset)kuduSession.close()})kuduClient.close()})})stream.foreachRDD(rdd => {val sqlProxy = new SqlProxy()val client = DataSourceUtil.getConnectiontry {val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRangesfor (or <- offsetRanges) {sqlProxy.executeUpdate(client, "replace into `offset_manager` (groupid,topic,`partition`,untilOffset) values(?,?,?,?)",Array(groupid, or.topic, or.partition.toString, or.untilOffset))}} catch {case e: Exception => e.printStackTrace()} finally {sqlProxy.shutdown(client)}})ssc.start()ssc.awaitTermination()}
}

5.2.5 效果展示

业务为根据设备实时统计count数需要基于历史数据,控制速度每秒1000条

数据无积压

impala查询展示

6 APACHE KUDU常见问题分析

6.1 APACHE KUDU服务启动失败

启动失败：

[zhouchen@hadoop102 ~]$ sudo service kudu-master start
Failed to start Kudu Master Server. Return value: 1        [失败]

[zhouchen@hadoop102 ~]$ ssh hadoop103 sudo service kudu-tserver start
Failed to start Kudu Tablet Server. Return value: 1        [失败]

[zhouchen@hadoop102 ~]$ ssh hadoop104 sudo service kudu-tserver start
Failed to start Kudu Tablet Server. Return value: 1        [失败]

检查ntp服务：

[zhouchen@hadoop102 ~]$ ntptime
ntp_gettime() returns code 5 (ERROR)time e2b95039.ced65000  Wed, Jul 15 2020 17:51:53.807, (.807958),maximum error 16000000 us, estimated error 16000000 us, TAI offset 0
ntp_adjtime() returns code 5 (ERROR)modes 0x0 (),offset 0.000 us, frequency -7.097 ppm, interval 1 s,maximum error 16000000 us, estimated error 16000000 us,status 0x41 (PLL,UNSYNC),time constant 7, precision 1.000 us, tolerance 500 ppm,

问题原因是ntp时间同步失败，重启ntp服务让时间同步正常：

[zhouchen@hadoop102 ~]$ xcall sudo service ntpd stop
要执行的命令是：sudo service ntpd stop
---------------------hadoop102------------------
关闭 ntpd：[确定]
---------------------hadoop103------------------
关闭 ntpd：[确定]
---------------------hadoop104------------------
关闭 ntpd：[确定]
[zhouchen@hadoop102 ~]$ xcall sudo service ntpd start
要执行的命令是：sudo service ntpd start
---------------------hadoop102------------------
正在启动 ntpd：[确定]
---------------------hadoop103------------------
正在启动 ntpd：[确定]
---------------------hadoop104------------------
正在启动 ntpd：[确定]
[zhouchen@hadoop102 ~]$ xcall sudo service ntpd status
要执行的命令是：sudo service ntpd status
---------------------hadoop102------------------
ntpd (pid  6051) 正在运行...
---------------------hadoop103------------------
ntpd (pid  5084) 正在运行...
---------------------hadoop104------------------
ntpd (pid  5081) 正在运行...

再次检查ntp服务：

[zhouchen@hadoop102 ~]$ xcall sudo ntptime
要执行的命令是：sudo ntptime
---------------------hadoop102------------------
ntp_gettime() returns code 0 (OK)time e2b95067.a7b794f0  Wed, Jul 15 2020 17:52:39.655, (.655145271),maximum error 8056622 us, estimated error 983 us, TAI offset 0
ntp_adjtime() returns code 0 (OK)modes 0x0 (),offset -2686.243 us, frequency -7.102 ppm, interval 1 s,maximum error 8056622 us, estimated error 983 us,status 0x2001 (PLL,NANO),time constant 6, precision 0.001 us, tolerance 500 ppm,
---------------------hadoop103------------------
ntp_gettime() returns code 0 (OK)time e2b95067.e1fc6d50  Wed, Jul 15 2020 17:52:39.882, (.882758322),maximum error 8145031 us, estimated error 10565 us, TAI offset 0
ntp_adjtime() returns code 0 (OK)modes 0x0 (),offset 28850.633 us, frequency -2.867 ppm, interval 1 s,maximum error 8145031 us, estimated error 10565 us,status 0x2001 (PLL,NANO),time constant 6, precision 0.001 us, tolerance 500 ppm,
---------------------hadoop104------------------
ntp_gettime() returns code 0 (OK)time e2b95068.272ce888  Wed, Jul 15 2020 17:52:40.153, (.153029112),maximum error 1078291 us, estimated error 10442 us, TAI offset 0
ntp_adjtime() returns code 0 (OK)modes 0x0 (),offset -29306.464 us, frequency -2.542 ppm, interval 1 s,maximum error 1078291 us, estimated error 10442 us,status 0x2001 (PLL,NANO),time constant 6, precision 0.001 us, tolerance 500 ppm,

启动Kudu成功:

[zhouchen@hadoop102 ~]$ sudo service kudu-master start
Started Kudu Master Server (kudu-master):                  [确定]