hadoop - Cassandra での hadoop mapreduce ジョブの結果が間違っている

Question

7 ノードの cassandra (1.1.1) と hadoop (1.03) クラスターがあります (タスクトラッカーはすべての cassandra ノードに同じようにインストールされます)。

私の列ファミリーは広い行パターンを使用しています。1 行には約 20 万列 (最大約 30 万列) が含まれます。

私の問題は、Hadoop を使用して分析ジョブを実行するとき (単語の出現回数をカウントする)、受け取った結果が間違っていることです (テスト レコードで予想したように、結果が低すぎます)。

ジョブ トラッカーで監視しているときに奇妙な点が 1 つあります。これは、マップの進行状況タスクが間違っていることを示しています (下の私の画像)。ジョブを再実行したときの「マップ入力レコード」の数 (同じデータ) は同じではありません。

ここに私の初期ジョブコードがあります:

    Job job = new Job(conf);
    job.setJobName(this.jobname);
    job.setJarByClass(BannerCount.class);

    job.setMapperClass(BannerViewMapper.class);
    job.setReducerClass(BannerClickReducer.class);
    FileSystem fs = FileSystem.get(conf);

    ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
    ConfigHelper.setInputInitialAddress(job.getConfiguration(), "192.168.23.114,192.168.23.115,192.168.23.116,192.168.23.117,192.168.23.121,192.168.23.122,192.168.23.123");
    ConfigHelper.setInputPartitioner(job.getConfiguration(), "org.apache.cassandra.dht.RandomPartitioner");
    ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY, true);
    ConfigHelper.setRangeBatchSize(job.getConfiguration(), 500);


    SlicePredicate predicate = new SlicePredicate();
    SliceRange sliceRange = new SliceRange();
    sliceRange.setStart(ByteBufferUtil.EMPTY_BYTE_BUFFER);
    sliceRange.setFinish(ByteBufferUtil.EMPTY_BYTE_BUFFER);
    sliceRange.setCount(200000);
    predicate.setSlice_range(sliceRange);
    ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);

    String outPathString = "BannerViewResultV3" + COLUMN_FAMILY;
    if (fs.exists(new Path(outPathString)))
        fs.delete(new Path(outPathString), true);

    FileOutputFormat.setOutputPath(job, new Path(outPathString));

    job.setInputFormatClass(ColumnFamilyInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);

    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(LongWritable.class);

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(LongWritable.class);
    job.setNumReduceTasks(28);
    job.waitForCompletion(true);
    return 1;