1

DSE Spark Job Serverを使用しています。私が達成しようとしているタスクは次のとおりです。

Java で作成したスパーク ジョブは、cassandra db からいくつかのデータをフェッチすることが期待されており、これは DSE Analytics クラスターにデプロイされます。

コードは次のとおりです。

package com.symantec.nsp.analytics;

import static com.datastax.spark.connector.japi.CassandraJavaUtil.javaFunctions;
import static com.datastax.spark.connector.japi.CassandraJavaUtil.mapRowTo;

import java.io.Serializable;
import java.util.List;
import java.util.UUID;

import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;

import spark.jobserver.JavaSparkJob;
import spark.jobserver.SparkJobInvalid;
import spark.jobserver.SparkJobValid$;
import spark.jobserver.SparkJobValidation;

import com.symantec.nsp.analytics.model.Bucket;
import com.typesafe.config.Config;

public class JavaSparkJobBasicQuery extends JavaSparkJob {

    public String runJob(JavaSparkContext jsc, Config config) {
        try {
            List<UUID> bucketRecords = javaFunctions(jsc).cassandraTable("nsp_storage", "bucket", mapRowTo(Bucket.class))
                    .select("id", "deleted").filter(s -> s.getDeleted()).map(s -> s.getId()).collect();

            System.out.println(">>>>>>>>  Total Buckets getting scanned by Spark :" + bucketRecords.size());
            return bucketRecords.toString();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    public SparkJobValidation validate(SparkContext sc, Config config) {
        return null;
    }

    public String invalidate(JavaSparkContext jsc, Config config) {
        return null;
    }
}

問題：

このコードを実行しているときに、次の問題が発生します。

  "status": "ERROR",
  "result":
    "message": "null",
    "errorClass": "scala.MatchError",
    "stack": ["spark.jobserver.JobManagerActor$$anonfun$spark$jobserver$JobManagerActor$$getJobFuture$4.apply(JobManagerActor.scala:244)", "scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)", "scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)", "java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)", "java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)", "java.lang.Thread.run(Thread.java:745)"]

誰かが問題を解決できますか。注：/tmpフォルダーのクリーニングを複数回試みました。この問題を解決できません。私が使用している DSE のバージョンは 4.8.10 です。

4

1 に答える 1

0

例外で null を返したくないかどうかはよくわかりません。私はそれを伝播するために残します。

于 2016-10-22T04:35:35.127 に答える