0

Cassandra クラスターで単純な mapred ジョブを実行していますが、出力をテーブルに保存しようとすると、InvalidRequestException(why:Column timestamp required) が発生します。

「タイムスタンプ」列をCFに手動で追加しようとしましたが、違いはありません。

これが私のCFの説明です（cqlshによって解釈されたとおり）：

CREATE TABLE output_words (
  key text PRIMARY KEY,
  "count" int,
) WITH COMPACT STORAGE AND
  bloom_filter_fp_chance=0.010000 AND
  caching='KEYS_ONLY' AND
  comment='' AND
  dclocal_read_repair_chance=0.000000 AND
  gc_grace_seconds=864000 AND
  read_repair_chance=0.100000 AND
  replicate_On_write='true' AND
  populate_io_cache_on_flush='false' AND
  compaction={'class': 'SizeTieredCompactionStrategy'} AND
  compression={'sstable_compression': 'SnappyCompressor'};

Cassandra v1.2.4の上にhadoop-core v1.1.2およびcassandra-thrift v1.2.4でPOMを使用しています

誰でもこれを回避する方法を提案できますか?

追加情報

次のようにジョブを構成しています（出力に関連する構成のみを表示しています）：

Job job = new Job(getConf(), "wordcount");

job.setJarByClass(TestJob.class);
job.setMapperClass(TokenizerMapper.class);
job.setReducerClass(ReducerToCassandra.class);

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(ByteBuffer.class);
job.setOutputValueClass(List.class);

job.setOutputFormatClass(ColumnFamilyOutputFormat.class);

ConfigHelper.setOutputColumnFamily(job.getConfiguration(), _keyspace, OUTPUT_COLUMN_FAMILY);

ConfigHelper.setOutputRpcPort(job.getConfiguration(), _port);
ConfigHelper.setOutputInitialAddress(job.getConfiguration(), _host);
ConfigHelper.setOutputPartitioner(job.getConfiguration(), "org.apache.cassandra.dht.Murmur3Partitioner");

そして私のレデューサークラス：

public static class ReducerToCassandra extends Reducer<Text, IntWritable, ByteBuffer, List<Mutation>>
{
    public void reduce(Text word, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(StringSerializer.get().toByteBuffer(word.toString()), Collections.singletonList(getMutation(word, sum)));
    }

    private static Mutation getMutation(Text word, int sum) {
        Column c = new Column();
        c.name = StringSerializer.get().toByteBuffer("count");
        c.value = IntegerSerializer.get().toByteBuffer(sum);
        c.timestamp = System.currentTimeMillis() * 1000;

        Mutation m = new Mutation();
        m.column_or_supercolumn = new ColumnOrSuperColumn();
        m.column_or_supercolumn.column = c;
        return m;
    }
}
4

1 に答える 1

1

これの代わりに

c.timestamp = System.currentTimeMillis() * 1000;

これを試すことができますか

c.setTimestamp(System.currentTimeMillis() * 1000)
于 2013-05-16T10:15:07.147 に答える