java - Spark-Java-foreachPartition

Question

我的环境如下 Spark 1.6.1Hadoop 2.6.2

我们的需求如下(全部在Java Spark中) 1. 读取 CSV 文件并应用架构并将其转换为数据框 2.通过Spark分区SQL获取所有不同的分区数据并并行迭代

当我尝试第二步时，我收到错误。

    DataFrame  sourceRowDF = hiveContext.createDataFrame(srcRowsRDD,source_Schema);
    sourceRowDF.registerTempTable("sourcetable");
    DataFrame  partitionedData = hiveContext.sql("select * from sourcetable distribute by region");

    partitionedData.foreachPartition( new ForeachPartitionFunc() {
    @Override
    public void call(Iterator<Row> it) {
        while (it.hasNext()) {
            System.out.println(it.next().toString());
        }
    }
});

抽象函数如下

public abstract class ForeachPartitionFunc extends AbstractFunction1<Iterator<Row>, BoxedUnit> implements Serializable {
        @Override
    public BoxedUnit apply(Iterator<Row> it) {
        call(it);
        return BoxedUnit.UNIT;
    }
    public abstract void call( Iterator<Row> it);
}

当我尝试编译此代码时，出现以下错误

Error:(254, 39) java: non-static variable this cannot be referenced from a static context

请问我做错了什么。

谢谢

巴拉

Answer 1

我必须在抽象实现中有一个公共(public)静态类。我更改了核心抽象实现如下

public  static class ForeachPartitionFunc extends AbstractFunction1<Iterator<Row>, BoxedUnit> implements Serializable {
        @Override
    public BoxedUnit apply(Iterator<Row> it) {
        call(it);
        return BoxedUnit.UNIT;
    }

    public void call(Iterator<Row> r) {
      System.out.println("this is to test");
    }
}