在scala Spark2中运行saveAsNewAPIHadoopDataset到hbase时出现空指针异常

我正在使用 saveAsNewAPIHadoopDataset 将 RDD 保存到 Hbase 中。以下是我的工作创建和提交。

    val outputTableName = "test3"
    val conf2 = HBaseConfiguration.create()
    conf2.set("hbase.zookeeper.quorum", "xx.xx.xx.xx")
    conf2.set("hbase.mapred.outputtable", outputTableName)
    conf2.set("mapreduce.outputformat.class", "org.apache.hadoop.hbase.mapreduce.TableOutputFormat")

    val job = createJob(outputTableName, conf2)
    val outputTable = sc.broadcast(outputTableName)
    val hbasePuts = simpleRdd.map(k => convertToPut(k, outputTable))

    hbasePuts.saveAsNewAPIHadoopDataset(job.getConfiguration)

这是我的创造就业功能

def createJob(table: String, conf: Configuration): Job = {
    conf.set(TableOutputFormat.OUTPUT_TABLE, table)
    val job = Job.getInstance(conf, this.getClass.getName.split('$')(0))
    job.setOutputFormatClass(classOf[TableOutputFormat[String]])
    job
  }

该函数将数据转换为Hbase格式

def convertToPut(k: (String, String, String), outputTable: Broadcast[String]): (ImmutableBytesWritable, Put) = {
    val rowkey = k._1
    val put = new Put(Bytes.toBytes(rowkey))
    val one = Bytes.toBytes("cf1")
    val two = Bytes.toBytes("cf2")

    put.addColumn(one, Bytes.toBytes("a"), Bytes.toBytes(k._2))
    put.addColumn(two, Bytes.toBytes("a"), Bytes.toBytes(k._3))
    (new ImmutableBytesWritable(Bytes.toBytes(outputTable.value)), put)
  }

这是我在第 125 行遇到的错误，即：hbasePuts.saveAsNewAPIHadoopDataset(job.getConfiguration)

Exception in thread "main" java.lang.NullPointerException
    at org.apache.hadoop.hbase.security.UserProvider.instantiate(UserProvider.java:122)
    at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:214)
    at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:119)
    at org.apache.hadoop.hbase.mapreduce.TableOutputFormat.checkOutputSpecs(TableOutputFormat.java:177)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1.apply$mcV$sp(PairRDDFunctions.scala:1099)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1.apply(PairRDDFunctions.scala:1085)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1.apply(PairRDDFunctions.scala:1085)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
    at org.apache.spark.rdd.PairRDDFunctions.saveAsNewAPIHadoopDataset(PairRDDFunctions.scala:1085)
    at ScalaSpark$.main(ScalaSpark.scala:125)
    at ScalaSpark.main(ScalaSpark.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:738)
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:187)
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:212)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:126)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

我也遇到过同样的问题。 我认为 org.apache.hadoop.hbase.mapreduce.TableOutputFormat 类中有一个错误。

TableOutputFormat原始代码如下：

public void checkOutputSpecs(JobContext context) throws IOException,
        InterruptedException {

    try (Admin admin = ConnectionFactory.createConnection(getConf()).getAdmin()) {
        TableName tableName = TableName.valueOf(this.conf.get(OUTPUT_TABLE));
        if (!admin.tableExists(tableName)) {
            throw new TableNotFoundException("Can't write, table does not exist:" +
                    tableName.getNameAsString());
        }

        if (!admin.isTableEnabled(tableName)) {
            throw new TableNotEnabledException("Can't write, table is not enabled: " +
                    tableName.getNameAsString());
        }
    }
}

如果我按如下方式修复它：

public void checkOutputSpecs(JobContext context) throws IOException,
        InterruptedException {

    //set conf by context parameter
    setConf(context.getConfiguration());

    try (Admin admin = ConnectionFactory.createConnection(getConf()).getAdmin()) {
        TableName tableName = TableName.valueOf(this.conf.get(OUTPUT_TABLE));
        if (!admin.tableExists(tableName)) {
            throw new TableNotFoundException("Can't write, table does not exist:" +
                    tableName.getNameAsString());
        }

        if (!admin.isTableEnabled(tableName)) {
            throw new TableNotEnabledException("Can't write, table is not enabled: " +
                    tableName.getNameAsString());
        }
    }
}

我的问题解决了。

Spark.hadoop.validateOutputSpecs

另一种解决方案是转Spark.hadoop.validateOutputSpecs https://spark.apache.org/docs/3.0.0/configuration.html#execution-behavior创建时关闭SparkSession.

val session = SparkSession.builder()
  .config("spark.hadoop.validateOutputSpecs", false)
  .getOrCreate()