文章目录
- 概要
- Kudu与Impala整合配置
- Impala内部表
- Impala外部表
- Impala sql操作kudu
- Impala jdbc操作表
- 如果使用了Hadoop 使用了Kerberos认证,可使用如下方式进行连接。
概要
- Impala是一个开源的高效率的SQL查询引擎,用于查询存储在Hadoop分布式文件系统(HDFS)中的大规模数据集。它提供了一个类似于传统关系型数据库的SQL接口,允许用户使用SQL语言查询存储在Hadoop集群中的数据。使用内存进行计算提供实时的SQL查询,impala强依赖于Hive 的MetaStore,直接使用hive的元数据,意味着impala元数据都存储在hive的MetaStore当中,并且impala兼容hive的绝大多数sql语法,具有实时,批处理,多并发等优点。
- Kudu提供了KuduClient api用于操作kudu数据库,但不支持标准SQL操作,可以将Kudu与Apache Impala紧密集成,impala天然就支持兼容kudu,允许开发人员使用Impala的SQL语法从Kudu的tablets 插入,查询,更新和删除数据,Kudu与Impala整合本质上就是为了可以使用Hive表来操作Kudu,主要支持SQL操作。
Kudu与Impala整合配置
先安装Impala后安装Kudu,Impala默认与Kudu没有形成依赖,这里需要首先在Impala中开启Kudu依赖支持,打开Impala->“配置”->“Kudu服务”:
以上配置完成之后,重启Impala即可。
Impala内部表
内部表是由Impala自身管理的表,数据存储在Hive元数据库和Kudu中。当删除内部表时,存储在Hive元数据库中的元数据和存储在kudu中的数据都会被删除。
例如:
CREATE TABLE my_table1
(
id BIGINT,
name STRING,
PRIMARY KEY(id)
)
PARTITION BY HASH PARTITIONS 16
STORED AS KUDU
TBLPROPERTIES(
‘kudu.master_addresses’ = ‘cm1:7051’,
‘kudu.table_name’ = ‘my_table1’
);
Impala外部表
外部表则是由KUDU管理的表,元数据存储在Hive元数据库中,但实际数据文件存储在kudu中。删除外部表时,只会删除元数据,实际的数据文件不会被删除。外部表也可以指定数据的存储位置,可以在建表时指定,也可以通过ALTER TABLE语句修改。
使用Kudu client api 在Kudu中创建表test_user,创建好之后。使用下面的sql语句创建外部表。
CREATE EXTERNAL TABLE test_user
STORED AS KUDU
TBLPROPERTIES(
‘kudu.table_name’ = ‘test_user’,
‘kudu.master_addresses’ = ‘10.68.18.60:7051’);
Impala sql操作kudu
插入数据
insert into default_vals(id,name,address,age) values (10,“hello1”,‘山东’,22) ;
查询表数据
select * from default_vals;
更新表数据
upsert into default_vals(id,name,address,age) values(102,‘hello2’,‘山东’,22);
删除数据
delete from default_valswhere id = 20;
Impala jdbc操作表
maven 依赖
<!-- impala的驱动 -->
<dependency>
<groupId>com.cloudera.impala.jdbc</groupId>
<artifactId>ImpalaJDBC42</artifactId>
<version>2.5.42</version>
<scope>provided</scope>
</dependency>
代码示例
package com.example.demo.impala;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
public class ImpalaCrud {
public static void main(String[] args) {
System.out.println("begin");
Connection conn =getConnection();
queryTable(conn) ;
// insertTable2(conn) ;
}
public static void insertTable2(Connection conn) {
String insertSql="insert into default_vals( name,age,create_time,update_time,id) values (?,?,?,now(),?)";
PreparedStatement ps=null;
try {
ps=conn.prepareStatement(insertSql);
ps.setString(1, "张三李四");
ps.setString(2, "43");
ps.setTimestamp(3, getCurrentTimestamp());
ps.setString(4, "102");
ps.execute();
} catch (SQLException e) {
e.printStackTrace();
}finally {
if(conn!=null) {
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}
}
public static java.sql.Timestamp getCurrentTimestamp() {
java.util.Date date=new java.util.Date();
java.sql.Timestamp timestamp = new java.sql.Timestamp(date.getTime());
System.out.println(timestamp);
// java.sql.Date sqlDate=new java.sql.Date(date.getTime());
return timestamp;
}
public static void insertTable(Connection conn) {
String insertSql="insert into default_vals( name,age,create_time,update_time,id) values (?,?,now(),now(),?)";
PreparedStatement ps=null;
try {
ps=conn.prepareStatement(insertSql);
ps.setString(1, "xxxxx1");
ps.setInt(2, 43);
ps.setInt(3, 101);
// ps.setInt(4, 33);
ps.execute();
} catch (SQLException e) {
e.printStackTrace();
}finally {
if(conn!=null) {
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}
}
public static void queryTable(Connection conn) {
String querySql="select * from test_user1";
// PreparedStatement ps=conn.prepareStatement(querySql);
Statement st;
try {
st = conn.createStatement();
ResultSet rs=st.executeQuery(querySql);
while(rs.next()) {
System.out.print (rs.getString(1));
System.out.print (rs.getString(2));
System.out.print (rs.getString(3));
System.out.println (" ");
}
rs.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally {
if(conn!=null) {
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}
}
public static Connection getConnection() {
Connection conn =null;
try {
Class.forName("com.cloudera.impala.jdbc.Driver");//指定连接类型
String url="jdbc:impala://10.68.18.170:21050/db1;UseSasl=0;AuthMech=0;UID=impala";
// String url="jdbc:impala://10.3.4.31:21050/ccit_dl_ods";
// conn = DriverManager.getConnection(url);//获取连接
conn = DriverManager.getConnection(url,"root","huawei@123");//获取连接
}catch(Exception e) {
e.printStackTrace();
}
return conn;
}
}
如果使用了Hadoop 使用了Kerberos认证,可使用如下方式进行连接。
package com.example.demo.impala;
import java.security.PrivilegedAction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import org.apache.hadoop.security.UserGroupInformation;
public class ImpalaKdc {
private static String driver = "com.cloudera.impala.jdbc.Driver";
public static void main(String[] args) throws Exception {
String jdbcUrl="jdbc:impala://cm2:21050/db1;AuthMech=1;KrbRealm=EXAMPLE.COM;KrbHostFQDN=cm2.cdh;KrbServiceName=impala";
String configPath="E:\\tmp\\krb5.conf";
String keyTabPath="E:\\tmp\\impala.keytab";
Connection conn=getImapalaAuthConnection(jdbcUrl,"impala/cm2.cdh",configPath,keyTabPath);
System.out.println(conn);
queryTable(conn);
}
private static Connection getImapalaAuthConnection(String jdbcUrl,String username,String configPath,String keyTabPath)throws Exception{
// System.setProperty("java.security.krb5.conf", configPath);
Connection connection = null;
try{
org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
conf.set("hadoop.security.authentication", "Kerberos");
UserGroupInformation.setConfiguration(conf);
UserGroupInformation.loginUserFromKeytab(username, keyTabPath);
connection = UserGroupInformation.getLoginUser().doAs(new PrivilegedAction<Connection>(){
@Override
public Connection run(){
Connection connection = null;
try{
Class.forName(driver);
connection = DriverManager.getConnection(jdbcUrl);
}catch (Exception e){
e.printStackTrace();
}
return connection;
}
});
}catch (Exception e){
throw e;
}
return connection;
}
public static void queryTable(Connection conn) {
String querySql="select * from test_user1";
// PreparedStatement ps=conn.prepareStatement(querySql);
Statement st;
try {
st = conn.createStatement();
ResultSet rs=st.executeQuery(querySql);
while(rs.next()) {
System.out.print (rs.getString(1));
System.out.print (rs.getString(2));
System.out.print (rs.getString(3));
System.out.println (" ");
}
rs.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally {
if(conn!=null) {
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}
}
}