Apache Spark与Cassandra行为

Question

我正在编写一个从Cassandra获取数据的独立Spark程序。我遵循示例并通过newAPIHadoopRDD（）和ClonFamilyInputFormat类创建了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());
}

例外情况：

JAVAio。NotSerializableException：java。nio。java上的HeapByteBuffer。io。ObjectOutputStream。java上的writeObject0（ObjectOutputStream.java:1164）。io。ObjectOutputStream。java上的defaultWriteFields（ObjectOutputStream.java:1518）。io。ObjectOutputStream。在java上编写SerialData（ObjectOutputStream.java:1483）。io。ObjectOutputStream。java上的writeOrdinaryObject（ObjectOutputStream.java:1400）。io。ObjectOutputStream。java上的writeObject0（ObjectOutputStream.java:1158）。io。ObjectOutputStream。组织中的writeObject（ObjectOutputStream.java:330）。阿帕奇。火花序列化程序。JavaSerializationStream。org上的writeObject（JavaSerializer.scala:42）。阿帕奇。火花存储DiskBlockObjectWriter。在org上编写（BlockObjectWriter.scala:179）。阿帕奇。火花调度程序。ShuffleMapTask$$anonfun$runTask$1。在组织中应用（ShuffleMapTask.scala:161）。阿帕奇。火花调度程序。ShuffleMapTask$$anonfun$runTask$1。在scala上应用（ShuffleMapTask.scala:158）。收集迭代器$类。foreach（迭代器.scala:727）位于org。阿帕奇。火花中断迭代器。foreach（interruptableiterator.scala:28）位于org。阿帕奇。火花调度程序。ShuffleMapTask。org上的runTask（ShuffleMapTask.scala:158）。阿帕奇。火花调度程序。ShuffleMapTask。运行任务（ShuffleMapTask.scala:99），位于组织。阿帕奇。火花调度程序。任务在组织上运行（Task.scala:51）。阿帕奇。火花执行人。执行者$TaskRunner。在java上运行（Executor.scala:187）。util。同时发生的ThreadPoolExecutor$工作者。java上的runTask（ThreadPoolExecutor.java:895）。util。同时发生的ThreadPoolExecutor$工作者。在java上运行（ThreadPoolExecutor.java:918）。lang.Thread。运行（Thread.java:662）

我要做的是将所有行键列合并到一个条目中。当我尝试使用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

有人知道问题出在哪里吗？序列化失败的原因是什么？

谢谢，
Shai

Answer 1

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

您应该尝试官方DataStax Cassandra driver for Spark，此处提供