SparkBulkload(Java)

在未用Bulkload写Hbase时&＃xff0c;使用RDD进行封装为Tuple2的KVRDD&＃xff0c;然后通过saveAsNewAPIHadoopDataset写Hbase&＃xff0c;非常慢&＃xff0c;400G的数据大概写了2H&＃43;还没写完&＃xff0c;后面没有办法就考虑使用Bulkload来导入数据。
在测试之前网上很多资料都是Scala版本的&＃xff0c;并且实现都是单个列来操作&＃xff0c;实际生产中会存在多个列族和列的情况&＃xff0c;并且这里面有很多坑。
先上代码&＃xff1a;

public class HbaseSparkUtils {private static Configuration hbaseConf;static {hbaseConf &＃61; HBaseConfiguration.create();hbaseConf.set(ConfigUtils.getHbaseZK()._1(), ConfigUtils.getHbaseZK()._2());hbaseConf.set(ConfigUtils.getHbaseZKPort()._1(), ConfigUtils.getHbaseZKPort()._2());}public static void saveHDFSHbaseHFile(SparkSession spark, // spark session Dataset ds, // 数据集String table_name, //hbase表名Integer rowKeyIndex, //rowkey的索引idString fields) throws Exception { // 数据集的字段列表hbaseConf.setInt("hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily", 1024);hbaseConf.set(TableOutputFormat.OUTPUT_TABLE, table_name);Job job &＃61; Job.getInstance();job.setMapOutputKeyClass(ImmutableBytesWritable.class);job.setMapOutputValueClass(KeyValue.class);job.setOutputFormatClass(HFileOutputFormat2.class);Connection conn &＃61; ConnectionFactory.createConnection(hbaseConf);TableName tableName &＃61; TableName.valueOf(table_name);HRegionLocator regionLocator &＃61; new HRegionLocator(tableName, (ClusterConnection) conn);Table realTable &＃61; ((ClusterConnection) conn).getTable(tableName);HFileOutputFormat2.configureIncrementalLoad(job, realTable, regionLocator);JavaRDD javaRDD &＃61; ds.toJavaRDD();JavaPairRDD javaPairRDD &＃61;javaRDD.mapToPair(new PairFunction>>() {&＃64;Overridepublic Tuple2>> call(Row row) throws Exception {List> tps &＃61; new ArrayList<>();String rowkey &＃61; row.getString(rowKeyIndex);ImmutableBytesWritable writable &＃61; new ImmutableBytesWritable(Bytes.toBytes(rowkey));// sort columns。这里需要对列进行排序&＃xff0c;不然会报错ArrayList> tuple2s &＃61; new ArrayList<>();String[] columns &＃61; fields.split(",");for (int i &＃61; 0; i (i, columns[i]));}for (Tuple2 t : tuple2s) {String[] fieldNames &＃61; row.schema().fieldNames();// 不将作为rowkey的字段存在列里面if (t._2().equals(fieldNames[rowKeyIndex])) {System.out.println(String.format("%s &＃61;&＃61; %s continue", t._2(), fieldNames[rowKeyIndex]));continue;}if ("main".equals(t._2())) {continue;}String value &＃61; getRowValue(row, t._1(), tuple2s.size());KeyValue kv &＃61; new KeyValue(Bytes.toBytes(rowkey),Bytes.toBytes(ConfigUtils.getFamilyInfo()._2()),Bytes.toBytes(t._2()), Bytes.toBytes(value));tps.add(new Tuple2<>(writable, kv));}for (Tuple2 t : tuple2s) {String value &＃61; getRowValue(row, t._1(), tuple2s.size());if ("main".equals(t._2())) { // filed &＃61;&＃61; &＃39;main&＃39;KeyValue kv &＃61; new KeyValue(Bytes.toBytes(rowkey),Bytes.toBytes(ConfigUtils.getFamilyMain()._2()),Bytes.toBytes(t._2()), Bytes.toBytes(value));tps.add(new Tuple2<>(writable, kv));break;}}return new Tuple2<>(writable, tps);}// 这里一定要按照rowkey进行排序&＃xff0c;这个效率很低&＃xff0c;目前没有找到优化的替代方案}).sortByKey().flatMapToPair(new PairFlatMapFunction>>,ImmutableBytesWritable, KeyValue>() {&＃64;Overridepublic Iterator> call(Tuple2>> tuple2s) throws Exception {return tuple2s._2().iterator();}});// 创建HDFS的临时HFile文件目录String temp &＃61; "/tmp/bulkload/"&＃43;table_name&＃43;"_"&＃43;System.currentTimeMillis();javaPairRDD.saveAsNewAPIHadoopFile(temp, ImmutableBytesWritable.class,KeyValue.class, HFileOutputFormat2.class, job.getConfiguration());LoadIncrementalHFiles loader &＃61; new LoadIncrementalHFiles(hbaseConf);Admin admin &＃61; conn.getAdmin();loader.doBulkLoad(new Path(temp), admin, realTable, regionLocator);}
}

2、下面是一些遇到的异常问题

源码分析&＃xff1a;