2

我正在尝试使用 Spark 从 Cassandra 读取数据。

    DataFrame rdf = sqlContext.read().option("keyspace", "readypulse")
            .option("table", "ig_posts")
            .format("org.apache.spark.sql.cassandra").load();

    rdf.registerTempTable("cassandra_table");
    System.out.println(sqlContext.sql("select count(external_id) from cassandra_table").collect()[0].getLong(0));

任务失败并出现以下错误。我无法理解为什么调用 ShuffleMaptask 以及为什么将其转换为 Task 是一个问题。

16/03/30 02:27:15 WARN TaskSetManager: Lost task 1.0 in stage 0.0 (TID 1, ip-10-165-180-22.ec2.internal): 
            java.lang.ClassCastException: 
            org.apache.spark.scheduler.ShuffleMapTask 
                cannot be cast to org.apache.spark.scheduler.Task
                        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:193)
                        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
                        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
                        at java.lang.Thread.run(Thread.java:745)
16/03/30 02:27:15 INFO TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0) on executor ip-10-165-180-22.ec2.internal: 
            java.lang.ClassCastException (org.apache.spark.scheduler.Shuf
                fleMapTask 
            cannot be cast to org.apache.spark.scheduler.Task) [duplicate 1]

16/03/30 02:27:15 ERROR TaskSetManager: Task 0 in stage 0.0 failed 4 times; aborting job

我正在使用 EMR 4.4、Spark 1.6、Cassandra 2.2（Datastax 社区）和 spark-cassandra-connector-java_2.10 1.6.0-M1（也尝试过 1.5.0）

我也尝试了以下代码，但得到了同样的错误。

    CassandraJavaRDD<CassandraRow> cjrdd = functions.cassandraTable(
            KEYSPACE, tableName).select(columns);
    logger.info("Got rows from cassandra " + cjrdd.count());

    JavaRDD<Double> jrdd2 = cjrdd.map(new Function<CassandraRow, Double>() {
        @Override
        public Double call(CassandraRow trainingRow) throws Exception {
            Object fCount = trainingRow.getRaw("follower_count");
            double count = 0;
            if (fCount != null) {
                count = (Long) fCount;
            }
            return count;
        }
    });
    logger.info("Mapper done : " + jrdd2.count());
    logger.info("Mapper done values : " + jrdd2.collect());
4

2 回答 2

1

最近由于 --conf spark.executor.userClassPathFirst=true.

引用Spark 的官方文档：

spark.executor.userClassPathFirst（实验）与 相同的功能spark.driver.userClassPathFirst，但应用于执行器实例。

我认为这些异常是由于一些 jar 版本冲突，并且根据 spark 文档，“用户的 jar 不应该包含 Hadoop 或 Spark 库，但是，这些将在运行时添加。”

于 2016-03-30T12:42:02.673 回答
0

我也在为同样的错误而苦苦挣扎。但是设置 userClassPathFirst=true 对我没有帮助。我设置了“spark.driver.userClassPathFirst=false”和“spark.executor.userClassPathFirst=false”。它解决了我的 ClassCastException 问题。这是我的命令。

spark-submit --conf "spark.driver.userClassPathFirst=false" --conf "spark.executor.userClassPathFirst=false" --deploy-mode client --class org.sdrc.kspmis.dashboardservice.KspDashboardServiceApplication --master spark: //192.168.1.95:7077 仪表板-0.0.1-SNAPSHOT-shaded.jar

于 2018-12-07T10:07:11.313 回答