在 Apache Spark 中，将 JavaRDD<Row> 转换为 Dataset<Row> 会出现异常：ArrayList is not a valid external type for string of schema

Question

我正在使用 hbase-spark connector 将 hbase 数据提取到 spark JavaRDD<Row> 中（我觉得我能够成功完成，因为我能够打印提取的 hbase 数据）。然后，我尝试将 JavaRDD<Row> 转换为 Dataset<Row>。但这给了我错误，该错误将在帖子中进一步给出。首先让我开始我的代码的样子。

private static JavaRDD<Row> loadHBaseRDD() throws ParseException
{
    //form list of row keys
    List<byte[]> rowKeys = new ArrayList<byte[]>(5);
    //consider ids is class level variable
    ids.forEach(id -> {
        rowKeys.add(Bytes.toBytes(id));     
    });
    JavaRDD<byte[]> rdd = jsc.parallelize(rowKeys);

    //make hbase-spark connector call 
    JavaRDD resultJRDD = jhbc.bulkGet(TableName.valueOf("table1"),2,rdd,new GetFunction(),new ResultFunction());

    return resultJRDD;
}

请注意，bulkGet() 接受实例 GetFunction 和 RsultFunction 类。 GetFunction 类具有返回Get 类实例的单一方法（来自hbase client）：

public static class GetFunction implements Function<byte[], Get> {
    private static final long serialVersionUID = 1L;
    public Get call(byte[] v) throws Exception {
        return new Get(v);
    }
}

ResultFunction 具有将Result（hbase 客户端类）的实例转换为Row 的函数：

public static class ResultFunction implements Function<Result, Row> 
{
    private static final long serialVersionUID = 1L;
    public Row call(Result result) throws Exception 
    {
        List<String> values = new ArrayList<String>(); //notice this is arraylist, we talk about this latter

        for (Cell cell : result.rawCells()) {
            values.add(Bytes.toString(CellUtil.cloneValue(cell)));
        }
        return RowFactory.create(values);
    }
}

当我调用loadHBaseRDD() 并打印返回值时，它会正确打印值：

JavaRDD<Row> hbaseJavaRDD = loadHBaseRDD();
hbaseJavaRDD.foreach(row -> { 
    System.out.println(row);   //this prints rows correctly
}); 

这意味着行已正确地从 hbase 获取到 spark。 现在我想将JavaRDD<Row> 转换为Dataset<Row>，如here 所述。因此我首先创建StructType：

StructType schema = //create schema

然后我尝试将JavaRDD 转换为数据框：

Dataset<Row> hbaseDataFrame = sparksession1.createDataFrame(hbaseJavaRDD, schema);
hbaseDataFrame.show(false);

这会在hbaseDataFrame.show(false) 行引发非常大的堆栈跟踪（下面仅显示一部分）异常，第一行如下：

java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: java.util.ArrayList is not a valid external type for schema of string

似乎，因为values 在ResultFunction.call() 中属于ArrayList 类型，所以它给出了异常java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: java.util.ArrayList is not a valid external type for schema of string。

stackoveflow 上有 [类似问题]，其中有 answer 说应该返回 String[][] 而不是列表。虽然我不明白返回String[][] 背后的原因，但我将ResultFunction 修改为values 类型为String[][]：

public static class ResultFunction implements Function<Result, Row> 
{
    private static final long serialVersionUID = 1L;
    public Row call(Result result) throws Exception 
    {
        String[] values = new String[result.rawCells().length];
        String[][] valuesWrapped = new String[1][]; 

        for(int i=0;i<result.rawCells().length;i++)
        {
            values[i] = Bytes.toString(CellUtil.cloneValue(result.rawCells()[i]));
        }
        valuesWrapped[0] = values;
        return RowFactory.create(valuesWrapped);
    }
}

它在同一行 hbaseDataFrame.show(false) 给出以下异常：

java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: [[Ljava.lang.String; is not a valid external type for schema of string

最后我再次修改了ResultFunction 类，使其具有values 类型为String[] 的变量：

public static class ResultFunction implements Function<Result, Row>
{
    private static final long serialVersionUID = 1L;
    public Row call(Result result) throws Exception 
    {
        String[] values = new String[result.rawCells().length];     
        for(int i=0;i<result.rawCells().length;i++)
        {
            values[i] = Bytes.toString(CellUtil.cloneValue(result.rawCells()[i]));
        }
        return values;
    }
}

这给了我一个例外，大堆栈跟踪有起始行：

java.lang.RuntimeException: Error while encoding: java.lang.ArrayIndexOutOfBoundsException: 14

那么这里可能出了什么问题？我应该怎么做？

Answer 1

最后一种方法（返回String[] values）是正确的。问题在于不正确的模式。似乎我以某种方式最终在架构中比数据中的列多了一列。 （感谢模式字符串中包含由单个空格分隔的列的额外空格字符。额外的空间正在创建额外的列。）