1

最近、時系列データをとして保持する1つの列ファミリーで学習するために4ノードのCassandraクラスターをセットアップしました。

キー->{列名：timeUUID、列値：csvログ行、ttl：1年}、NetflixAstyanaxJavaクライアントを使用して約100万のログ行を読み込みます。

また、Cassandraデータに対していくつかの分析を実行するために、1つのnamenodeと4つのdatanodeでmap-reduceジョブを実行するようにHadoopを構成しました。

インターネットで利用可能なすべての例では、Hadoopジョブ構成のSlicePredicateとして列名を使用していますが、列としてtimeUUIDがあるので、一度に1000列のバッチでCassandraデータをHadoopジョブ構成ツールに効率的にフィードするにはどうすればよいですか。

このテストデータの一部の行には10000を超える列があり、実際のデータではさらに多くなると予想されます。

私は自分の仕事を次のように構成します

public int run(String[] arg0) throws Exception {
    Job job = new Job(getConf(), JOB_NAME);
Job.setJarByClass(LogTypeCounterByDate.class);
job.setMapperClass(LogTypeCounterByDateMapper.class);       
job.setReducerClass(LogTypeCounterByDateReducer.class);

job.setInputFormatClass(ColumnFamilyInputFormat.class);

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

ConfigHelper.setRangeBatchSize(getConf(), 1000);


SliceRange sliceRange = new SliceRange(ByteBuffer.wrap(new byte[0]), 
    ByteBuffer.wrap(new byte[0]), true, 1000);

SlicePredicate slicePredicate = new SlicePredicate();
slicePredicate.setSlice_range(sliceRange);


ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
ConfigHelper.setInputRpcPort(job.getConfiguration(), INPUT_RPC_PORT);
ConfigHelper.setInputInitialAddress(job.getConfiguration(), INPUT_INITIAL_ADRESS);
    ConfigHelper.setInputPartitioner(job.getConfiguration(), INPUT_PARTITIONER);
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), slicePredicate);
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
job.waitForCompletion(true);
return job.isSuccessful() ? 0 : 1;
}

しかし、私がMapperをどのように定義しているか理解できません。親切に、Mapperクラスのテンプレートを提供していただけませんか。

public static class LogTypeCounterByDateMapper extends Mapper<ByteBuffer, SortedMap<ByteBuffer, IColumn>, Text, LongWritable>
{
    private Text key = null;
    private LongWritable value = null;

    @Override
    protected void setup(Context context){

    }

    public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns, Context context){
        //String[] lines = columns.;

    }
}
4

1 に答える 1

0 
ConfigHelper.setRangeBatchSize(getConf(), 1000)
...
SlicePredicate predicate = new SlicePredicate().setSlice_range(new SliceRange(TimeUUID.asByteBuffer(startValue), TimeUUID.asByteBuffer(endValue), false, 1000))
ConfigHelper.setInputSlicePredicate(conf, predicate)
于 2012-11-12T21:55:37.703 に答える