1

我正在运行带有 2 个 Worker 的 spark 集群，每个 60GB。

我为 JdbcRDD 编写了以下代码。

String sql   = "SELECT * FROM( SELECT ROW_NUMBER() OVER(ORDER BY (SELECT 1)) AS Row,"+ 
               " * FROM [Table_1]) A WHERE Row >= ? AND Row <= ? ";

    SparkContext sctx   = new SparkContext(getSparkConf());
    try {
        JdbcRDD<List> jdbcRdd = new JdbcRDD(sctx,new GetJDBCConnection(),sql,0, rowCount, 200, new GetJDBCResult(),scala.reflect.ClassTag$.MODULE$.AnyRef());

        Object[] bb = (Object[])jdbcRdd.collect();

        System.out.println("Length of Object array : "+bb.length);
        System.out.println("JdbcRDD:- "+bb);
    } catch(Exception e) {
        e.printStackTrace();
    }

GetJdbcResult 的代码是

class GetJDBCResult extends AbstractFunction1<ResultSet, List> implements Serializable{

private static final long serialVersionUID = -78825308090L;
 public List apply(ResultSet rs) {
    Object result = null;

    List lst = new ArrayList();
    try {
        System.out.println("In apply method");
        System.out.println("resultSet : -"+rs);
        int cols  = rs.getMetaData().getColumnCount();
        System.out.println("no of columns : "+cols);
        for(int i=1;i<=cols;i++) {
            result = rs.getObject(i);

            System.out.println("Object : -"+result);
            lst.add(result);
        }
        System.out.println("result->" + lst);
    } catch (SQLException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return lst;         
}

}

上面的代码运行良好，我在独立模式（本地 *）下运行 Spark，但如果使用集群环境，则会抛出以下错误：

org.apache.spark.SparkException: Job aborted due to stage failure: Task 7 in stage 0.0 failed 4 times, most recent failure: Lost task 7.3 in stage 0.0 (TID 39, DD1AI7511): java.lang.NullPointerException: 
    org.apache.spark.rdd.JdbcRDD$$anon$1.<init>(JdbcRDD.scala:74)
    org.apache.spark.rdd.JdbcRDD.compute(JdbcRDD.scala:70)
    org.apache.spark.rdd.JdbcRDD.compute(JdbcRDD.scala:50)
    org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:262)
    org.apache.spark.rdd.RDD.iterator(RDD.scala:229)
    org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
    org.apache.spark.scheduler.Task.run(Task.scala:54)
    org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
    java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
    java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
    java.lang.Thread.run(Thread.java:722)

驱动程序堆栈跟踪：

Worker 日志上没有任何跟踪/日志。我在这里做错了吗？有人知道吗？

4

1 回答 1

1

如果您查看 JdbcRDD 中的第 74 行，很明显这是由于数据库连接为空。

https://github.com/apache/spark/blob/655699f8b7156e8216431393436368e80626cdb2/core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala

由于它在本地工作但不在集群中，请检查您的集群实例是否可以访问数据库。如果这是在 EC2 中，请确保您的防火墙规则是正确的。还要确保 spark 集群在与您的数据库相同的 VPC 中运行。验证这一点的最佳方法是通过 SSH 连接到您的一个 spark 从节点，并查看您是否可以从那里远程连接到数据库。

于 2014-12-29T07:59:04.423 回答