4

我正在编写一个从 Cassandra 获取数据的独立 Spark 程序。我按照示例并通过 newAPIHadoopRDD() 和 ColumnFamilyInputFormat 类创建了 RDD。创建了 RDD，但是当我调用 RDD 的 .groupByKey() 方法时，我得到了 NotSerializableException：

public static void main(String[] args) {
    SparkConf sparkConf = new SparkConf();
    sparkConf.setMaster("local").setAppName("Test");
    JavaSparkContext ctx = new JavaSparkContext(sparkConf);

    Job job = new Job();
    Configuration jobConf = job.getConfiguration();
    job.setInputFormatClass(ColumnFamilyInputFormat.class);

    ConfigHelper.setInputInitialAddress(jobConf, host);
    ConfigHelper.setInputRpcPort(jobConf, port);
    ConfigHelper.setOutputInitialAddress(jobConf, host);
    ConfigHelper.setOutputRpcPort(jobConf, port);
    ConfigHelper.setInputColumnFamily(jobConf, keySpace, columnFamily, true);
    ConfigHelper.setInputPartitioner(jobConf,"Murmur3Partitioner");
    ConfigHelper.setOutputPartitioner(jobConf,"Murmur3Partitioner");

    SlicePredicate predicate = new SlicePredicate();
    SliceRange sliceRange = new SliceRange();
    sliceRange.setFinish(new byte[0]);
    sliceRange.setStart(new byte[0]);
    predicate.setSlice_range(sliceRange);
    ConfigHelper.setInputSlicePredicate(jobConf, predicate);

    JavaPairRDD<ByteBuffer, SortedMap<ByteBuffer, IColumn>> rdd =
    spark.newAPIHadoopRDD(jobConf,
    ColumnFamilyInputFormat.class.asSubclass(org.apache.hadoop.mapreduce.InputFormat.class),
    ByteBuffer.class, SortedMap.class);

    JavaPairRDD<ByteBuffer, Iterable<SortedMap<ByteBuffer, IColumn>>> groupRdd = rdd.groupByKey();
    System.out.println(groupRdd.count());
}

例外：

java.io.NotSerializableException：在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1164) 在 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1518) 在 java.io.ObjectOutputStream.writeSerialData( ObjectOutputStream.java:1483) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1400) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1158) 在 java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:330) ) 在 org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42) 在 org.apache.spark.storage.DiskBlockObjectWriter.write(BlockObjectWriter.scala:179) 在 org.apache.spark.scheduler.ShuffleMapTask$ $anonfun$runTask$1.apply(ShuffleMapTask.scala:161) 在 org.apache.spark.scheduler。ShuffleMapTask$$anonfun$runTask$1.apply(ShuffleMapTask.scala:158) at scala.collection.Iterator$class.foreach(Ite​​rator.scala:727) at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)在 org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:158) 在 org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) 在 org.apache.spark.scheduler.Task.run (Task.scala:51) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:187) at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:895) at java。 util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:918) 在 java.lang.Thread.run(Thread.java:662)InterruptibleIterator.foreach(InterruptibleIterator.scala:28) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:158) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org .apache.spark.scheduler.Task.run(Task.scala:51) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:187) at java.util.concurrent.ThreadPoolExecutor$Worker.runTask (ThreadPoolExecutor.java:895) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:918) 在 java.lang.Thread.run(Thread.java:662)InterruptibleIterator.foreach(InterruptibleIterator.scala:28) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:158) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org .apache.spark.scheduler.Task.run(Task.scala:51) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:187) at java.util.concurrent.ThreadPoolExecutor$Worker.runTask (ThreadPoolExecutor.java:895) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:918) 在 java.lang.Thread.run(Thread.java:662)executor.Executor$TaskRunner.run(Executor.scala:187) at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:895) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:第918章）executor.Executor$TaskRunner.run(Executor.scala:187) at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:895) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:第918章）

我想要做的是将所有行键列合并到一个条目中。当我尝试像这样使用 reduceByKey() 方法时，我也会遇到同样的异常：

JavaPairRDD<ByteBuffer, SortedMap<ByteBuffer, IColumn>> reducedRdd = rdd.reduceByKey(
    new Function2<SortedMap<ByteBuffer, IColumn>, SortedMap<ByteBuffer, IColumn>, sortedMap<ByteBuffer, IColumn>>() {
        public SortedMap<ByteBuffer, IColumn> call(SortedMap<ByteBuffer, IColumn> arg0,
            SortedMap<ByteBuffer, IColumn> arg1) throws Exception {
            SortedMap<ByteBuffer, IColumn> sortedMap = new TreeMap<ByteBuffer, IColumn>(arg0.comparator());
            sortedMap.putAll(arg0);
            sortedMap.putAll(arg1);
            return sortedMap;
        }
    }
);

我在用：

  • spark-1.0.0-bin-hadoop1
  • 卡桑德拉 1.2.12
  • Java 1.6

有谁知道问题是什么？是什么导致序列化失败？

谢谢，
夏

4

1 回答 1

4

您的问题可能是由于尝试序列化 ByteBuffers 引起的。它们不可序列化，您需要在生成 RDD 之前将它们转换为字节数组。

您应该尝试用于 Spark 的官方 DataStax Cassandra 驱动程序，该驱动程序可在此处获得

于 2014-07-02T13:39:29.163 回答