# java样例代码 #### 开发说明 本样例只适用于MRS的HBase。 - 前提条件 在DLI管理控制台上已完成创建跨源连接并绑定队列。具体操作请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)》。 ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png) 认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。 - 代码实现 1. 导入依赖 - 涉及到的mvn依赖库 ``` org.apache.spark spark-sql_2.11 2.3.2 ``` - import相关依赖包 ``` import org.apache.spark.sql.SparkSession; ``` 2. 创建会话 ``` parkSession = SparkSession.builder().appName("datasource-HBase-MRS").getOrCreate(); ``` - 通过SQL API 访问 - 未开启Kerberos认证 1. 创建DLI跨源访问MRS HBase的关联表，填写连接参数。 ``` sparkSession.sql("CREATE TABLE testhbase(id STRING, location STRING, city STRING) using hbase OPTIONS('ZKHost'='10.0.0.63:2181','TableName'='hbtest','RowKey'='id:5','Cols'='location:info.location,city:detail.city') "); ``` 2. 插入数据 ``` sparkSession.sql("insert into testhbase values('12345','abc','xxx')"); ``` 3. 查询数据 ``` sparkSession.sql("select * from testhbase").show(); ``` 插入数据后： ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001135051618.png) - 开启Kerberos认证 1. 创建DLI跨源访问MRS HBase的关联表，填写连接参数。 ``` sparkSession.sql("CREATE TABLE testhbase(id STRING, location STRING, city STRING) using hbase OPTIONS('ZKHost'='10.0.0.63:2181','TableName'='hbtest','RowKey'='id:5','Cols'='location:info.location,city:detail.city,'krb5conf'='./krb5.conf','keytab'='./user.keytab','principal'='krbtest') "); ``` 与未开启kerberos认证相比，开启了kerberos认证需要多设置三个参数，如[表1]所示。 表1参数说明 | 参数名称与参数值 | 参数说明 | |:---|:---| | 'krb5conf' = './krb5.conf' | krb5.conf的地址。 | | 'keytab'='./user.keytab' | Keytab的地址。 | | 'principal' ='krbtest' | 认证用户名。 | krb5.conf和keytab文件获取请具体参考[开启Kerberos认证时的相关配置文件](https://support.huaweicloud.com/devg-dli/dli_09_0196.html#dli_09_0196__section12676527182715)操作说明。 2. 插入数据 ``` sparkSession.sql("insert into testhbase values('95274','abc','Hongkong')"); ``` 3. 查询数据 ``` sparkSession.sql("select * from testhbase").show(); ``` - 提交Spark作业 1. 将写好的代码文件生成jar包，上传至OBS桶中。 2. 如果MRS集群开启了Kerberos认证，创建Spark作业时需要将krb5.conf和user.keytab文件添加到作业的依赖文件中，未开启Kerberos认证该步骤忽略。如[图1]所示： 图1添加依赖文件 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0000001186508057.png "点击放大") 3. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。 ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png) - 如果选择spark版本为2.3.2（即将下线）或2.4.5提交作业时，需要指定Module模块，名称为：sys.datasource.hbase。 - 如果选择Spark版本为3.1.1及以上版本时，无需选择Module模块， 需在 'Spark参数（--conf)' 配置 spark.driver.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/hbase/\* spark.executor.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/hbase/\* - 通过控制台提交作业请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0384.html)》中的"选择依赖资源参数说明"。 - 通过API提交作业请参考《数据湖探索API参考》\>《[创建批处理作业](https://support.huaweicloud.com/api-dli/dli_02_0124.html)》中"表2-请求参数说明"关于"modules"参数的说明。 #### 完整示例代码 - 通过SQL API访问 - **未开启Kerberos** 完整代码示例 ``` import org.apache.spark.sql.SparkSession; public class java_mrs_hbase { public static void main(String[] args) { //create a SparkSession session SparkSession sparkSession = SparkSession.builder().appName("datasource-HBase-MRS").getOrCreate(); sparkSession.sql("CREATE TABLE testhbase(id STRING, location STRING, city STRING) using hbase OPTIONS('ZKHost'='10.0.0.63:2181','TableName'='hbtest','RowKey'='id:5','Cols'='location:info.location,city:detail.city') "); //*****************************SQL model*********************************** sparkSession.sql("insert into testhbase values('95274','abc','Hongkong')"); sparkSession.sql("select * from testhbase").show(); sparkSession.close(); } } ``` - **开启Kerberos** 完整代码示例 ``` import org.apache.spark.SparkContext; import org.apache.spark.SparkFiles; import org.apache.spark.sql.SparkSession; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; public class Test_HBase_SparkSql_Kerberos { private static void copyFile(File src,File dst) throws IOException { InputStream input = null; OutputStream output = null; try { input = new FileInputStream(src); output = new FileOutputStream(dst); byte[] buf = new byte[1024]; int bytesRead; while ((bytesRead = input.read(buf)) > 0) { output.write(buf, 0, bytesRead); } } finally { input.close(); output.close(); } } public static void main(String[] args) throws InterruptedException, IOException { SparkSession sparkSession = SparkSession.builder().appName("Test_HBase_SparkSql_Kerberos").getOrCreate(); SparkContext sc = sparkSession.sparkContext(); sc.addFile("obs://xietest1/lzq/krb5.conf"); sc.addFile("obs://xietest1/lzq/user.keytab"); Thread.sleep(20); File krb5_startfile = new File(SparkFiles.get("krb5.conf")); File keytab_startfile = new File(SparkFiles.get("user.keytab")); String path_user = System.getProperty("user.dir"); File keytab_endfile = new File(path_user + "/" + keytab_startfile.getName()); File krb5_endfile = new File(path_user + "/" + krb5_startfile.getName()); copyFile(krb5_startfile,krb5_endfile); copyFile(keytab_startfile,keytab_endfile); Thread.sleep(20); /** * Create an association table for the DLI association Hbase table */ sparkSession.sql("CREATE TABLE testhbase(id string,booleanf boolean,shortf short,intf int,longf long,floatf float,doublef double) " + "using hbase OPTIONS(" + "'ZKHost'='10.0.0.146:2181'," + "'TableName'='hbtest'," + "'RowKey'='id:100'," + "'Cols'='booleanf:CF1.booleanf,shortf:CF1.shortf,intf:CF1.intf,longf:CF2.longf,floatf:CF1.floatf,doublef:CF2.doublef'," + "'krb5conf'='" + path_user + "/krb5.conf'," + "'keytab'='" + path_user+ "/user.keytab'," + "'principal'='krbtest') "); //*****************************SQL model*********************************** sparkSession.sql("insert into testhbase values('newtest',true,1,2,3,4,5)"); sparkSession.sql("select * from testhbase").show(); sparkSession.close(); } } ```