| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| package org.apache.kudu.mapreduce.tools; |
| |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.COLUMN_CLIENT; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.COLUMN_KEY_ONE; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.COLUMN_KEY_TWO; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.COLUMN_PREV_ONE; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.COLUMN_PREV_TWO; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.COLUMN_ROW_ID; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.COLUMN_UPDATE_COUNT; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.Counts; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.DEFAULT_HEADS_TABLE_NAME; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.DEFAULT_TABLE_NAME; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.HEADS_TABLE_NAME_KEY; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.TABLE_NAME_KEY; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.Xoroshiro128PlusRandom; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.getCreateTableOptions; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.getHeadsTableSchema; |
| import static org.apache.kudu.mapreduce.tools.BigLinkedListCommon.getTableSchema; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.UUID; |
| |
| import com.google.common.base.Joiner; |
| import com.google.common.collect.ImmutableList; |
| import org.apache.commons.cli.CommandLine; |
| import org.apache.commons.cli.GnuParser; |
| import org.apache.commons.cli.HelpFormatter; |
| import org.apache.commons.cli.Options; |
| import org.apache.commons.cli.ParseException; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.conf.Configured; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.io.BytesWritable; |
| import org.apache.hadoop.io.NullWritable; |
| import org.apache.hadoop.io.Text; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.mapreduce.Counter; |
| import org.apache.hadoop.mapreduce.CounterGroup; |
| import org.apache.hadoop.mapreduce.Counters; |
| import org.apache.hadoop.mapreduce.InputFormat; |
| import org.apache.hadoop.mapreduce.InputSplit; |
| import org.apache.hadoop.mapreduce.Job; |
| import org.apache.hadoop.mapreduce.JobContext; |
| import org.apache.hadoop.mapreduce.Mapper; |
| import org.apache.hadoop.mapreduce.RecordReader; |
| import org.apache.hadoop.mapreduce.Reducer; |
| import org.apache.hadoop.mapreduce.TaskAttemptContext; |
| import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; |
| import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; |
| import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; |
| import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat; |
| import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; |
| import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; |
| import org.apache.hadoop.util.Tool; |
| import org.apache.hadoop.util.ToolRunner; |
| import org.apache.yetus.audience.InterfaceAudience; |
| import org.apache.yetus.audience.InterfaceStability; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.kudu.Schema; |
| import org.apache.kudu.client.AbstractKuduScannerBuilder; |
| import org.apache.kudu.client.Bytes; |
| import org.apache.kudu.client.CreateTableOptions; |
| import org.apache.kudu.client.KuduClient; |
| import org.apache.kudu.client.KuduException; |
| import org.apache.kudu.client.KuduScanner; |
| import org.apache.kudu.client.KuduSession; |
| import org.apache.kudu.client.KuduTable; |
| import org.apache.kudu.client.Operation; |
| import org.apache.kudu.client.PartialRow; |
| import org.apache.kudu.client.RowResult; |
| import org.apache.kudu.client.RowResultIterator; |
| import org.apache.kudu.client.SessionConfiguration; |
| import org.apache.kudu.client.Update; |
| import org.apache.kudu.mapreduce.CommandLineParser; |
| import org.apache.kudu.mapreduce.KuduTableMapReduceUtil; |
| import org.apache.kudu.util.Pair; |
| |
| /** |
| * <p> |
| * This is an integration test borrowed from goraci, written by Keith Turner, |
| * which is in turn inspired by the Accumulo test called continuous ingest (ci). |
| * The original source code can be found here: |
| * </p> |
| * <ul> |
| * <li> |
| * <a href="https://github.com/keith-turner/goraci">https://github.com/keith-turner/goraci</a> |
| * </li> |
| * <li> |
| * <a href="https://github.com/enis/goraci/">https://github.com/enis/goraci/</a> |
| * </li> |
| * </ul> |
| * |
| * <p> |
| * Apache Accumulo has a simple test suite that verifies that data is not |
| * lost at scale. This test suite is called continuous ingest. This test runs |
| * many ingest clients that continually create linked lists containing 25 |
| * million nodes. At some point the clients are stopped and a map reduce job is |
| * run to ensure no linked list has a hole. A hole indicates data was lost. |
| * </p> |
| * |
| * <p> |
| * The nodes in the linked list are random. This causes each linked list to |
| * spread across the table. Therefore if one part of a table loses data, then it |
| * will be detected by references in another part of the table. |
| * </p> |
| * |
| * <h3> |
| * THE ANATOMY OF THE TEST |
| * </h3> |
| * |
| * <p> |
| * Below is rough sketch of how data is written. For specific details look at |
| * the Generator code. |
| * </p> |
| * <ol> |
| * <li> |
| * Write out 1 million nodes |
| * </li> |
| * <li> |
| * Flush the client |
| * </li> |
| * <li> |
| * Write out 1 million that reference previous million |
| * </li> |
| * <li> |
| * If this is the 25th set of 1 million nodes, then update 1st set of million to point to last |
| * </li> |
| * <li> |
| * Goto 1 |
| * </li> |
| * </ol> |
| * |
| * <p> |
| * The key is that nodes only reference flushed nodes. Therefore a node should |
| * never reference a missing node, even if the ingest client is killed at any |
| * point in time. |
| * </p> |
| * |
| * <p> |
| * When running this test suite w/ Accumulo there is a script running in |
| * parallel called the Agitator that randomly and continuously kills server |
| * processes. The outcome was that many data loss bugs were found in Accumulo |
| * by doing this. This test suite can also help find bugs that impact uptime |
| * and stability when run for days or weeks. |
| * </p> |
| * |
| * <p> |
| * This test suite consists the following: |
| * </p> |
| * <ul> |
| * <li> |
| * A few Java programs |
| * </li> |
| * <li> |
| * A little helper script to run the java programs |
| * </li> |
| * <li> |
| * A maven script to build it. |
| * </li> |
| * </ul> |
| * |
| * <p> |
| * When generating data, its best to have each map task generate a multiple of |
| * 25 million. The reason for this is that circular linked list are generated |
| * every 25M. Not generating a multiple in 25M will result in some nodes in the |
| * linked list not having references. The loss of an unreferenced node can not |
| * be detected. |
| * </p> |
| * |
| * <h3> |
| * Below is a description of the Java programs |
| * </h3> |
| * |
| * <ul> |
| * <li> |
| * Generator - A map only job that generates data. As stated previously, |
| * its best to generate data in multiples of 25M. |
| * </li> |
| * <li> |
| * Verify - A map reduce job that looks for holes. Look at the counts after running. REFERENCED and |
| * UNREFERENCED are ok, any UNDEFINED counts are bad. Do not run at the same |
| * time as the Generator. |
| * </li> |
| * <li> |
| * Print - A standalone program that prints nodes in the linked list |
| * </li> |
| * <li> |
| * Delete - Disabled. A standalone program that deletes a single node |
| * </li> |
| * <li> |
| * Walker - Disabled. A standalone program that start following a linked list and emits timing |
| * info. |
| * </li> |
| * </ul> |
| * |
| * <h3> |
| * KUDU-SPECIFIC CHANGES |
| * </h3> |
| * |
| * <ul> |
| * <li> |
| * The 16 bytes row key is divided into two 8 byte long since we don't have a "bytes" type in |
| * Kudu. Note that the C++ client can store bytes directly in string columns. Using longs |
| * enables us to pretty print human readable keys than can then be passed back just as easily. |
| * </li> |
| * <li> |
| * The table can be pre-split when running the Generator. The row keys' first component will be |
| * spread over the Long.MIN_VALUE - Long.MAX_VALUE keyspace. |
| * </li> |
| * <li> |
| * The Walker and Deleter programs were disabled to save some time but they can be re-enabled then |
| * ported to Kudu without too much effort. |
| * </li> |
| * </ul> |
| */ |
| @InterfaceAudience.Private |
| @InterfaceStability.Unstable |
| public class IntegrationTestBigLinkedList extends Configured implements Tool { |
| private static final byte[] NO_KEY = new byte[1]; |
| |
| /** How many rows to write per map task. This has to be a multiple of 25M. */ |
| private static final String GENERATOR_NUM_ROWS_PER_MAP_KEY |
| = "IntegrationTestBigLinkedList.generator.num_rows"; |
| |
| private static final String GENERATOR_NUM_MAPPERS_KEY |
| = "IntegrationTestBigLinkedList.generator.map.tasks"; |
| |
| private static final String GENERATOR_WIDTH_KEY |
| = "IntegrationTestBigLinkedList.generator.width"; |
| |
| private static final String GENERATOR_WRAP_KEY |
| = "IntegrationTestBigLinkedList.generator.wrap"; |
| |
| private static final int WIDTH_DEFAULT = 1000000; |
| private static final int WRAP_DEFAULT = 25; |
| private static final int ROWKEY_LENGTH = 16; |
| |
| private String toRun; |
| private String[] otherArgs; |
| |
| static class CINode { |
| String key; |
| String prev; |
| String client; |
| long rowId; |
| int updateCount; |
| } |
| |
| /** |
| * A Map only job that generates random linked list and stores them. |
| */ |
| static class Generator extends Configured implements Tool { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(Generator.class); |
| |
| private CommandLineParser parser; |
| private KuduClient client; |
| |
| static class GeneratorInputFormat extends InputFormat<BytesWritable,NullWritable> { |
| |
| static class GeneratorInputSplit extends InputSplit implements Writable { |
| @Override |
| public long getLength() throws IOException, InterruptedException { |
| return 1; |
| } |
| |
| @Override |
| public String[] getLocations() throws IOException, InterruptedException { |
| return new String[0]; |
| } |
| |
| @Override |
| public void readFields(DataInput arg0) throws IOException { |
| } |
| |
| @Override |
| public void write(DataOutput arg0) throws IOException { |
| } |
| } |
| |
| static class GeneratorRecordReader extends RecordReader<BytesWritable,NullWritable> { |
| private long count; |
| private long numNodes; |
| private Xoroshiro128PlusRandom rand; |
| |
| @Override |
| public void close() throws IOException { |
| } |
| |
| @Override |
| public BytesWritable getCurrentKey() throws IOException, InterruptedException { |
| byte[] bytes = new byte[ROWKEY_LENGTH]; |
| rand.nextBytes(bytes); |
| return new BytesWritable(bytes); |
| } |
| |
| @Override |
| public NullWritable getCurrentValue() throws IOException, InterruptedException { |
| return NullWritable.get(); |
| } |
| |
| @Override |
| public float getProgress() throws IOException, InterruptedException { |
| return (float)(count / (double)numNodes); |
| } |
| |
| @Override |
| public void initialize(InputSplit arg0, TaskAttemptContext context) |
| throws IOException, InterruptedException { |
| numNodes = context.getConfiguration().getLong(GENERATOR_NUM_ROWS_PER_MAP_KEY, 25000000); |
| // Use SecureRandom to avoid issue described in HBASE-13382. |
| rand = new Xoroshiro128PlusRandom(); |
| } |
| |
| @Override |
| public boolean nextKeyValue() throws IOException, InterruptedException { |
| return count++ < numNodes; |
| } |
| } |
| |
| @Override |
| public RecordReader<BytesWritable,NullWritable> createRecordReader( |
| InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { |
| GeneratorRecordReader rr = new GeneratorRecordReader(); |
| rr.initialize(split, context); |
| return rr; |
| } |
| |
| @Override |
| public List<InputSplit> getSplits(JobContext job) throws IOException, InterruptedException { |
| int numMappers = job.getConfiguration().getInt(GENERATOR_NUM_MAPPERS_KEY, 1); |
| |
| ArrayList<InputSplit> splits = new ArrayList<InputSplit>(numMappers); |
| |
| for (int i = 0; i < numMappers; i++) { |
| splits.add(new GeneratorInputSplit()); |
| } |
| |
| return splits; |
| } |
| } |
| |
| /** Ensure output files from prev-job go to map inputs for current job */ |
| static class OneFilePerMapperSFIF<K, V> extends SequenceFileInputFormat<K, V> { |
| @Override |
| protected boolean isSplitable(JobContext context, Path filename) { |
| return false; |
| } |
| } |
| |
| /** |
| * Some ASCII art time: |
| * [ . . . ] represents one batch of random longs of length WIDTH |
| * |
| * _________________________ |
| * | ______ | |
| * | | || |
| * __+_________________+_____ || |
| * v v v ||| |
| * first = [ . . . . . . . . . . . ] ||| |
| * ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ||| |
| * | | | | | | | | | | | ||| |
| * prev = [ . . . . . . . . . . . ] ||| |
| * ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ||| |
| * | | | | | | | | | | | ||| |
| * current = [ . . . . . . . . . . . ] ||| |
| * ||| |
| * ... ||| |
| * ||| |
| * last = [ . . . . . . . . . . . ] ||| |
| * | | | | | | | | | | |-----||| |
| * | |--------|| |
| * |___________________________| |
| */ |
| static class GeneratorMapper |
| extends Mapper<BytesWritable, NullWritable, NullWritable, NullWritable> { |
| |
| private byte[][] first = null; |
| private byte[][] prev = null; |
| private byte[][] current = null; |
| private String id; |
| private long rowId = 0; |
| private int i; |
| private KuduClient client; |
| private KuduTable table; |
| private KuduSession session; |
| private KuduTable headsTable; |
| private long numNodes; |
| private long wrap; |
| private int width; |
| |
| @Override |
| protected void setup(Context context) throws KuduException { |
| id = "Job: " + context.getJobID() + " Task: " + context.getTaskAttemptID(); |
| Configuration conf = context.getConfiguration(); |
| CommandLineParser parser = new CommandLineParser(conf); |
| client = parser.getClient(); |
| table = client.openTable(getTableName(conf)); |
| headsTable = client.openTable(getHeadsTable(conf)); |
| session = client.newSession(); |
| session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH); |
| session.setMutationBufferSpace(WIDTH_DEFAULT); |
| session.setIgnoreAllDuplicateRows(true); |
| |
| this.width = context.getConfiguration().getInt(GENERATOR_WIDTH_KEY, WIDTH_DEFAULT); |
| current = new byte[this.width][]; |
| int wrapMultiplier = context.getConfiguration().getInt(GENERATOR_WRAP_KEY, WRAP_DEFAULT); |
| this.wrap = (long)wrapMultiplier * width; |
| this.numNodes = context.getConfiguration().getLong( |
| GENERATOR_NUM_ROWS_PER_MAP_KEY, (long)WIDTH_DEFAULT * WRAP_DEFAULT); |
| if (this.numNodes < this.wrap) { |
| this.wrap = this.numNodes; |
| } |
| } |
| |
| @Override |
| protected void cleanup(Context context) throws KuduException { |
| session.close(); |
| client.shutdown(); |
| } |
| |
| @Override |
| protected void map(BytesWritable key, NullWritable value, Context output) throws IOException { |
| current[i] = new byte[key.getLength()]; |
| System.arraycopy(key.getBytes(), 0, current[i], 0, key.getLength()); |
| if (++i == current.length) { |
| persist(output, current, false); |
| i = 0; |
| |
| // Keep track of the first row so that we can point to it at the end. |
| if (first == null) { |
| first = current; |
| } |
| prev = current; |
| current = new byte[this.width][]; |
| |
| rowId += current.length; |
| output.setStatus("Count " + rowId); |
| |
| // Check if it's time to wrap up this batch. |
| if (rowId % wrap == 0) { |
| // this block of code turns the 1 million linked list of length 25 into one giant |
| // circular linked list of 25 million. |
| circularLeftShift(first); |
| |
| persist(output, first, true); |
| |
| Operation insert = headsTable.newInsert(); |
| PartialRow row = insert.getRow(); |
| row.addLong(COLUMN_KEY_ONE, Bytes.getLong(first[0])); |
| row.addLong(COLUMN_KEY_TWO, Bytes.getLong(first[0], 8)); |
| session.apply(insert); |
| session.flush(); |
| |
| first = null; |
| prev = null; |
| } |
| } |
| } |
| |
| private static <T> void circularLeftShift(T[] first) { |
| T ez = first[0]; |
| System.arraycopy(first, 1, first, 0, first.length - 1); |
| first[first.length - 1] = ez; |
| } |
| |
| private void persist(Context output, byte[][] data, boolean update) throws KuduException { |
| for (int i = 0; i < data.length; i++) { |
| Operation put = update ? table.newUpdate() : table.newInsert(); |
| PartialRow row = put.getRow(); |
| |
| long keyOne = Bytes.getLong(data[i]); |
| long keyTwo = Bytes.getLong(data[i], 8); |
| |
| row.addLong(COLUMN_KEY_ONE, keyOne); |
| row.addLong(COLUMN_KEY_TWO, keyTwo); |
| |
| // prev is null for the first line, we'll update it at the end. |
| if (prev == null) { |
| row.setNull(COLUMN_PREV_ONE); |
| row.setNull(COLUMN_PREV_TWO); |
| } else { |
| row.addLong(COLUMN_PREV_ONE, Bytes.getLong(prev[i])); |
| row.addLong(COLUMN_PREV_TWO, Bytes.getLong(prev[i], 8)); |
| } |
| |
| if (!update) { |
| // We only add those for new inserts, we don't update the heads with a new row, etc. |
| row.addLong(COLUMN_ROW_ID, rowId + i); |
| row.addString(COLUMN_CLIENT, id); |
| row.addInt(COLUMN_UPDATE_COUNT, 0); |
| } |
| session.apply(put); |
| |
| if (i % 1000 == 0) { |
| // Tickle progress every so often else maprunner will think us hung |
| output.progress(); |
| } |
| } |
| |
| session.flush(); |
| } |
| } |
| |
| @Override |
| public int run(String[] args) throws Exception { |
| if (args.length < 4) { |
| System.out.println("Usage : " + Generator.class.getSimpleName() + |
| " <num mappers> <num nodes per map> <num_tablets> <tmp output dir> [<width> <wrap " + |
| "multiplier>]"); |
| System.out.println(" where <num nodes per map> should be a multiple of " + |
| " width*wrap multiplier, 25M by default"); |
| return 0; |
| } |
| |
| int numMappers = Integer.parseInt(args[0]); |
| long numNodes = Long.parseLong(args[1]); |
| int numTablets = Integer.parseInt(args[2]); |
| Path tmpOutput = new Path(args[3]); |
| Integer width = (args.length < 5) ? null : Integer.parseInt(args[4]); |
| Integer wrapMultiplier = (args.length < 6) ? null : Integer.parseInt(args[5]); |
| return run(numMappers, numNodes, numTablets, tmpOutput, width, wrapMultiplier); |
| } |
| |
| public int run(int numMappers, long numNodes, int numTablets, Path tmpOutput, |
| Integer width, Integer wrapMultiplier) throws Exception { |
| parser = new CommandLineParser(getConf()); |
| client = parser.getClient(); |
| try { |
| int ret = runRandomInputGenerator(numMappers, numNodes, tmpOutput, width, wrapMultiplier); |
| if (ret > 0) { |
| return ret; |
| } |
| return runGenerator(numMappers, numNodes, numTablets, tmpOutput, width, wrapMultiplier); |
| } finally { |
| client.close(); |
| client = null; |
| } |
| } |
| |
| |
| private void createTables(int numTablets) throws Exception { |
| createSchema(getTableName(getConf()), getTableSchema(), numTablets); |
| createSchema(getHeadsTable(getConf()), getHeadsTableSchema(), numTablets); |
| } |
| |
| private void createSchema(String tableName, Schema schema, int numTablets) throws Exception { |
| if (client.tableExists(tableName)) { |
| return; |
| } |
| |
| CreateTableOptions builder = getCreateTableOptions(schema, |
| parser.getNumReplicas(), |
| numTablets, 1); |
| client.createTable(tableName, schema, builder); |
| } |
| |
| public int runRandomInputGenerator(int numMappers, long numNodes, Path tmpOutput, |
| Integer width, Integer wrapMultiplier) throws Exception { |
| LOG.info("Running RandomInputGenerator with numMappers=" + numMappers + |
| ", numNodes=" + numNodes); |
| Job job = new Job(getConf()); |
| |
| job.setJobName("Random Input Generator"); |
| job.setNumReduceTasks(0); |
| job.setJarByClass(getClass()); |
| |
| job.setInputFormatClass(GeneratorInputFormat.class); |
| job.setOutputKeyClass(BytesWritable.class); |
| job.setOutputValueClass(NullWritable.class); |
| |
| setJobConf(job, numMappers, numNodes, width, wrapMultiplier); |
| |
| job.setMapperClass(Mapper.class); //identity mapper |
| |
| FileOutputFormat.setOutputPath(job, tmpOutput); |
| job.setOutputFormatClass(SequenceFileOutputFormat.class); |
| |
| boolean success = job.waitForCompletion(true); |
| |
| return success ? 0 : 1; |
| } |
| |
| public int runGenerator(int numMappers, long numNodes, int numTablets, Path tmpOutput, |
| Integer width, Integer wrapMultiplier) throws Exception { |
| LOG.info("Running Generator with numMappers=" + numMappers + ", numNodes=" + numNodes); |
| createTables(numTablets); |
| |
| Job job = new Job(getConf()); |
| |
| job.setJobName("Link Generator"); |
| job.setNumReduceTasks(0); |
| job.setJarByClass(getClass()); |
| |
| FileInputFormat.setInputPaths(job, tmpOutput); |
| job.setInputFormatClass(org.apache.kudu.mapreduce.tools.IntegrationTestBigLinkedList |
| .Generator.OneFilePerMapperSFIF.class); |
| job.setOutputKeyClass(NullWritable.class); |
| job.setOutputValueClass(NullWritable.class); |
| |
| setJobConf(job, numMappers, numNodes, width, wrapMultiplier); |
| |
| job.setMapperClass(GeneratorMapper.class); |
| |
| job.setOutputFormatClass(NullOutputFormat.class); |
| |
| job.getConfiguration().setBoolean("mapred.map.tasks.speculative.execution", false); |
| // If we fail, retrying will fail again in case we were able to flush at least once since |
| // we'll be creating duplicate rows. Better to just have one try. |
| job.getConfiguration().setInt("mapreduce.map.maxattempts", 1); |
| // Lack of YARN-445 means we can't auto-jstack on timeout, so disabling the timeout gives |
| // us a chance to do it manually. |
| job.getConfiguration().setInt("mapreduce.task.timeout", 0); |
| |
| KuduTableMapReduceUtil.addDependencyJars(job); |
| KuduTableMapReduceUtil.addCredentialsToJob(client, job); |
| |
| boolean success = job.waitForCompletion(true); |
| |
| return success ? 0 : 1; |
| } |
| } |
| |
| /** |
| * A Map Reduce job that verifies that the linked lists generated by |
| * {@link Generator} do not have any holes. |
| */ |
| static class Verify extends Configured implements Tool { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(Verify.class); |
| private static final BytesWritable DEF = new BytesWritable(NO_KEY); |
| private static final Joiner COMMA_JOINER = Joiner.on(","); |
| private static final byte[] rowKey = new byte[ROWKEY_LENGTH]; |
| private static final byte[] prev = new byte[ROWKEY_LENGTH]; |
| |
| private Job job; |
| |
| public static class VerifyMapper extends Mapper<NullWritable, RowResult, |
| BytesWritable, BytesWritable> { |
| private BytesWritable row = new BytesWritable(); |
| private BytesWritable ref = new BytesWritable(); |
| |
| @Override |
| protected void map(NullWritable key, RowResult value, Mapper.Context context) |
| throws IOException, InterruptedException { |
| Bytes.setLong(rowKey, value.getLong(0)); |
| Bytes.setLong(rowKey, value.getLong(1), 8); |
| |
| row.set(rowKey, 0, rowKey.length); |
| // Emit that the row is defined |
| context.write(row, DEF); |
| if (value.isNull(2)) { |
| LOG.warn(String.format("Prev is not set for: %s", Bytes.pretty(rowKey))); |
| } else { |
| Bytes.setLong(prev, value.getLong(2)); |
| Bytes.setLong(prev, value.getLong(3), 8); |
| ref.set(prev, 0, prev.length); |
| // Emit which row is referenced by this row. |
| context.write(ref, row); |
| } |
| } |
| } |
| |
| public static class VerifyReducer extends Reducer<BytesWritable,BytesWritable,Text,Text> { |
| private ArrayList<byte[]> refs = new ArrayList<byte[]>(); |
| |
| @Override |
| public void reduce(BytesWritable key, Iterable<BytesWritable> values, Context context) |
| throws IOException, InterruptedException { |
| |
| int defCount = 0; |
| |
| refs.clear(); |
| // We only expect two values, a DEF and a reference, but there might be more. |
| for (BytesWritable type : values) { |
| if (type.getLength() == DEF.getLength()) { |
| defCount++; |
| } else { |
| byte[] bytes = new byte[type.getLength()]; |
| System.arraycopy(type.getBytes(), 0, bytes, 0, type.getLength()); |
| refs.add(bytes); |
| } |
| } |
| |
| // TODO check for more than one def, should not happen |
| |
| List<String> refsList = new ArrayList<>(refs.size()); |
| String keyString = null; |
| if (defCount == 0 || refs.size() != 1) { |
| for (byte[] ref : refs) { |
| refsList.add(COMMA_JOINER.join(Bytes.getLong(ref), Bytes.getLong(ref, 8))); |
| } |
| keyString = COMMA_JOINER.join(Bytes.getLong(key.getBytes()), |
| Bytes.getLong(key.getBytes(), 8)); |
| |
| LOG.error("Linked List error: Key = " + keyString + " References = " + refsList); |
| } |
| |
| if (defCount == 0 && refs.size() > 0) { |
| // this is bad, found a node that is referenced but not defined. It must have been |
| // lost, emit some info about this node for debugging purposes. |
| context.write(new Text(keyString), new Text(refsList.toString())); |
| context.getCounter(Counts.UNDEFINED).increment(1); |
| } else if (defCount > 0 && refs.size() == 0) { |
| // node is defined but not referenced |
| context.write(new Text(keyString), new Text("none")); |
| context.getCounter(Counts.UNREFERENCED).increment(1); |
| } else { |
| if (refs.size() > 1) { |
| if (refsList != null) { |
| context.write(new Text(keyString), new Text(refsList.toString())); |
| } |
| context.getCounter(Counts.EXTRAREFERENCES).increment(refs.size() - 1); |
| } |
| // node is defined and referenced |
| context.getCounter(Counts.REFERENCED).increment(1); |
| } |
| |
| } |
| } |
| |
| @Override |
| public int run(String[] args) throws Exception { |
| |
| if (args.length != 2) { |
| System.out.println("Usage : " + Verify.class.getSimpleName() + |
| " <output dir> <num reducers>"); |
| return 0; |
| } |
| |
| String outputDir = args[0]; |
| int numReducers = Integer.parseInt(args[1]); |
| |
| return run(outputDir, numReducers); |
| } |
| |
| public int run(String outputDir, int numReducers) throws Exception { |
| return run(new Path(outputDir), numReducers); |
| } |
| |
| public int run(Path outputDir, int numReducers) throws Exception { |
| LOG.info("Running Verify with outputDir=" + outputDir + ", numReducers=" + numReducers); |
| |
| job = new Job(getConf()); |
| |
| job.setJobName("Link Verifier"); |
| job.setNumReduceTasks(numReducers); |
| job.setJarByClass(getClass()); |
| |
| Joiner columnsToQuery = Joiner.on(","); |
| |
| new KuduTableMapReduceUtil.TableInputFormatConfiguratorWithCommandLineParser( |
| job, getTableName(getConf()), |
| columnsToQuery.join(COLUMN_KEY_ONE, COLUMN_KEY_TWO, COLUMN_PREV_ONE, COLUMN_PREV_TWO)) |
| .configure(); |
| job.setMapperClass(VerifyMapper.class); |
| job.setMapOutputKeyClass(BytesWritable.class); |
| job.setMapOutputValueClass(BytesWritable.class); |
| job.getConfiguration().setBoolean("mapred.map.tasks.speculative.execution", false); |
| |
| job.setReducerClass(VerifyReducer.class); |
| job.setOutputFormatClass(TextOutputFormat.class); |
| TextOutputFormat.setOutputPath(job, outputDir); |
| |
| boolean success = job.waitForCompletion(true); |
| |
| return success ? 0 : 1; |
| } |
| |
| @SuppressWarnings("deprecation") |
| public boolean verify(long expectedReferenced) throws Exception { |
| if (job == null) { |
| throw new IllegalStateException("You should call run() first"); |
| } |
| |
| Counters counters = job.getCounters(); |
| |
| Counter referenced = counters.findCounter(Counts.REFERENCED); |
| Counter unreferenced = counters.findCounter(Counts.UNREFERENCED); |
| Counter undefined = counters.findCounter(Counts.UNDEFINED); |
| Counter multiref = counters.findCounter(Counts.EXTRAREFERENCES); |
| |
| boolean success = true; |
| //assert |
| if (expectedReferenced != referenced.getValue()) { |
| LOG.error("Expected referenced count does not match with actual referenced count. " + |
| "Expected referenced=" + expectedReferenced + ", actual=" + referenced.getValue()); |
| success = false; |
| } |
| |
| if (unreferenced.getValue() > 0) { |
| boolean couldBeMultiRef = (multiref.getValue() == unreferenced.getValue()); |
| LOG.error("Unreferenced nodes were not expected. Unreferenced count=" + |
| unreferenced.getValue() + |
| (couldBeMultiRef ? "; could be due to duplicate random numbers" : "")); |
| success = false; |
| } |
| |
| if (undefined.getValue() > 0) { |
| LOG.error("Found an undefined node. Undefined count=" + undefined.getValue()); |
| success = false; |
| } |
| |
| // TODO Add the rows' location on failure. |
| if (!success) { |
| //Configuration conf = job.getConfiguration(); |
| //HConnection conn = HConnectionManager.getConnection(conf); |
| //TableName tableName = getTableName(conf); |
| CounterGroup g = counters.getGroup("undef"); |
| Iterator<Counter> it = g.iterator(); |
| while (it.hasNext()) { |
| String keyString = it.next().getName(); |
| //byte[] key = Bytes.toBytes(keyString); |
| //HRegionLocation loc = conn.relocateRegion(tableName, key); |
| LOG.error("undefined row " + keyString /*+ ", " + loc*/); |
| } |
| g = counters.getGroup("unref"); |
| it = g.iterator(); |
| while (it.hasNext()) { |
| String keyString = it.next().getName(); |
| //byte[] key = Bytes.toBytes(keyString); |
| //HRegionLocation loc = conn.relocateRegion(tableName, key); |
| LOG.error("unreferred row " + keyString /*+ ", " + loc*/); |
| } |
| } |
| return success; |
| } |
| } |
| |
| /** |
| * Executes Generate and Verify in a loop. Data is not cleaned between runs, so each iteration |
| * adds more data. |
| */ |
| static class Loop extends Configured implements Tool { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(Loop.class); |
| |
| IntegrationTestBigLinkedList it; |
| |
| FileSystem fs; |
| |
| protected void runGenerator(int numMappers, long numNodes, |
| int numTablets, String outputDir, |
| Integer width, Integer wrapMultiplier) throws Exception { |
| Path outputPath = new Path(outputDir); |
| UUID uuid = UUID.randomUUID(); //create a random UUID. |
| Path generatorOutput = new Path(outputPath, uuid.toString()); |
| |
| Generator generator = new Generator(); |
| generator.setConf(getConf()); |
| int retCode = generator.run(numMappers, numNodes, numTablets, generatorOutput, width, |
| wrapMultiplier); |
| if (retCode > 0) { |
| throw new RuntimeException("Generator failed with return code: " + retCode); |
| } |
| fs.delete(generatorOutput, true); |
| } |
| |
| protected void runVerify(String outputDir, |
| int numReducers, |
| long expectedNumNodes, |
| int retries) throws Exception { |
| // Kudu doesn't fully support snapshot consistency so we might start reading from a node that |
| // doesn't have all the data. This happens often with under "chaos monkey"-type of setups. |
| for (int i = 0; i < retries; i++) { |
| if (i > 0) { |
| long sleep = 60 * 1000; |
| LOG.info("Retrying in " + sleep + "ms"); |
| Thread.sleep(sleep); |
| } |
| |
| Path outputPath = new Path(outputDir); |
| UUID uuid = UUID.randomUUID(); //create a random UUID. |
| Path iterationOutput = new Path(outputPath, uuid.toString()); |
| |
| Verify verify = new Verify(); |
| verify.setConf(getConf()); |
| int retCode = verify.run(iterationOutput, numReducers); |
| if (retCode > 0) { |
| LOG.warn("Verify.run failed with return code: " + retCode); |
| } else if (!verify.verify(expectedNumNodes)) { |
| LOG.warn("Verify.verify failed"); |
| } else { |
| fs.delete(iterationOutput, true); |
| LOG.info("Verify finished with success. Total nodes=" + expectedNumNodes); |
| return; |
| } |
| } |
| throw new RuntimeException("Ran out of retries to verify"); |
| } |
| |
| @Override |
| public int run(String[] args) throws Exception { |
| if (args.length < 6) { |
| System.err.println("Usage: Loop <num iterations> <num mappers> <num nodes per mapper> " + |
| "<num_tablets> <output dir> <num reducers> [<width> <wrap multiplier>" + |
| "<start expected nodes> <num_verify_retries>]"); |
| return 1; |
| } |
| LOG.info("Running Loop with args:" + Arrays.deepToString(args)); |
| |
| int numIterations = Integer.parseInt(args[0]); |
| int numMappers = Integer.parseInt(args[1]); |
| long numNodes = Long.parseLong(args[2]); |
| int numTablets = Integer.parseInt(args[3]); |
| String outputDir = args[4]; |
| int numReducers = Integer.parseInt(args[5]); |
| Integer width = (args.length < 7) ? null : Integer.parseInt(args[6]); |
| Integer wrapMultiplier = (args.length < 8) ? null : Integer.parseInt(args[7]); |
| long expectedNumNodes = (args.length < 9) ? 0 : Long.parseLong(args[8]); |
| int numVerifyRetries = (args.length < 10) ? 3 : Integer.parseInt(args[9]); |
| |
| if (numIterations < 0) { |
| numIterations = Integer.MAX_VALUE; // run indefinitely (kind of) |
| } |
| |
| fs = FileSystem.get(getConf()); |
| |
| for (int i = 0; i < numIterations; i++) { |
| LOG.info("Starting iteration = " + i); |
| runGenerator(numMappers, numNodes, numTablets, outputDir, width, wrapMultiplier); |
| expectedNumNodes += numMappers * numNodes; |
| |
| runVerify(outputDir, numReducers, expectedNumNodes, numVerifyRetries); |
| } |
| |
| return 0; |
| } |
| } |
| |
| /** |
| * A stand alone program that prints out portions of a list created by {@link Generator} |
| */ |
| private static class Print extends Configured implements Tool { |
| @Override |
| public int run(String[] args) throws Exception { |
| Options options = new Options(); |
| options.addOption("s", "start", true, "start key, only the first component"); |
| options.addOption("e", "end", true, "end key (exclusive), only the first component"); |
| options.addOption("l", "limit", true, "number to print"); |
| |
| GnuParser parser = new GnuParser(); |
| CommandLine cmd = null; |
| try { |
| cmd = parser.parse(options, args); |
| if (cmd.getArgs().length != 0) { |
| throw new ParseException("Command takes no arguments"); |
| } |
| } catch (ParseException e) { |
| System.err.println("Failed to parse command line " + e.getMessage()); |
| System.err.println(); |
| HelpFormatter formatter = new HelpFormatter(); |
| formatter.printHelp(getClass().getSimpleName(), options); |
| System.exit(-1); |
| } |
| |
| CommandLineParser cmdLineParser = new CommandLineParser(getConf()); |
| long timeout = cmdLineParser.getOperationTimeoutMs(); |
| KuduClient client = cmdLineParser.getClient(); |
| |
| KuduTable table = client.openTable(getTableName(getConf())); |
| KuduScanner.KuduScannerBuilder builder = |
| client.newScannerBuilder(table) |
| .scanRequestTimeout(timeout); |
| |
| |
| if (cmd.hasOption("s")) { |
| PartialRow row = table.getSchema().newPartialRow(); |
| row.addLong(0, Long.parseLong(cmd.getOptionValue("s"))); |
| builder.lowerBound(row); |
| } |
| if (cmd.hasOption("e")) { |
| PartialRow row = table.getSchema().newPartialRow(); |
| row.addLong(0, Long.parseLong(cmd.getOptionValue("e"))); |
| builder.exclusiveUpperBound(row); |
| } |
| |
| int limit = cmd.hasOption("l") ? Integer.parseInt(cmd.getOptionValue("l")) : 100; |
| |
| int count = 0; |
| |
| KuduScanner scanner = builder.build(); |
| while (scanner.hasMoreRows() && count < limit) { |
| RowResultIterator rowResults = scanner.nextRows(); |
| count = printNodesAndGetNewCount(count, limit, rowResults); |
| } |
| RowResultIterator rowResults = scanner.close(); |
| printNodesAndGetNewCount(count, limit, rowResults); |
| |
| client.shutdown(); |
| |
| return 0; |
| } |
| |
| private static int printNodesAndGetNewCount(int oldCount, int limit, |
| RowResultIterator rowResults) { |
| int newCount = oldCount; |
| if (rowResults == null) { |
| return newCount; |
| } |
| |
| CINode node = new CINode(); |
| for (RowResult result : rowResults) { |
| newCount++; |
| node = getCINode(result, node); |
| printCINodeString(node); |
| if (newCount == limit) { |
| break; |
| } |
| } |
| return newCount; |
| } |
| } |
| |
| /** |
| * This tool needs to be run separately from the Generator-Verify loop. It can run while the |
| * other two are running or in between loops. |
| * |
| * Each mapper scans a "heads" table and, for each row, follows the circular linked list and |
| * updates their counter until it reaches the head of the list again. |
| */ |
| private static class Updater extends Configured implements Tool { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(Updater.class); |
| |
| private static final String MAX_LINK_UPDATES_PER_MAPPER = "kudu.updates.per.mapper"; |
| |
| public enum Counts { |
| // Stats on what we're updating. |
| UPDATED_LINKS, |
| UPDATED_NODES, |
| FIRST_UPDATE, |
| SECOND_UPDATE, |
| THIRD_UPDATE, |
| FOURTH_UPDATE, |
| MORE_THAN_FOUR_UPDATES, |
| // Stats on what's broken. |
| BROKEN_LINKS, |
| BAD_UPDATE_COUNTS |
| } |
| |
| public static class UpdaterMapper extends Mapper<NullWritable, RowResult, |
| NullWritable, NullWritable> { |
| private KuduClient client; |
| private KuduTable table; |
| private KuduSession session; |
| |
| /** |
| * Schema we use when getting rows from the linked list, we only need the reference and |
| * its update count. |
| */ |
| private static final List<String> SCAN_COLUMN_NAMES = |
| ImmutableList.of(COLUMN_PREV_ONE, COLUMN_PREV_TWO, COLUMN_UPDATE_COUNT, COLUMN_CLIENT); |
| |
| private int numUpdatesPerMapper; |
| |
| /** |
| * Processing each linked list takes minutes, meaning that it's easily possible for our |
| * scanner to timeout. Instead, we gather all the linked list heads that we need and |
| * process them all at once in the first map invocation. |
| */ |
| private List<Pair<Long, Long>> headsCache; |
| |
| @Override |
| protected void setup(Context context) throws KuduException { |
| Configuration conf = context.getConfiguration(); |
| CommandLineParser parser = new CommandLineParser(conf); |
| client = parser.getClient(); |
| table = client.openTable(getTableName(conf)); |
| session = client.newSession(); |
| numUpdatesPerMapper = conf.getInt(MAX_LINK_UPDATES_PER_MAPPER, 1); |
| headsCache = new ArrayList<>(numUpdatesPerMapper); |
| } |
| |
| @Override |
| protected void map(NullWritable key, RowResult value, Mapper.Context context) |
| throws IOException, InterruptedException { |
| // Add as many heads as we need, then we skip the rest. |
| do { |
| if (headsCache.size() < numUpdatesPerMapper) { |
| value = (RowResult)context.getCurrentValue(); |
| headsCache.add(new Pair<Long, Long>(value.getLong(0), value.getLong(1))); |
| } |
| } while (context.nextKeyValue()); |
| |
| // At this point we've exhausted the scanner and hopefully gathered all the linked list |
| // heads we needed. |
| LOG.info("Processing " + headsCache.size() + |
| " linked lists, out of " + numUpdatesPerMapper); |
| processAllHeads(context); |
| } |
| |
| private void processAllHeads(Mapper.Context context) throws IOException { |
| for (Pair<Long, Long> value : headsCache) { |
| processHead(value, context); |
| } |
| } |
| |
| private void processHead(Pair<Long, Long> head, Mapper.Context context) throws IOException { |
| long headKeyOne = head.getFirst(); |
| long headKeyTwo = head.getSecond(); |
| long prevKeyOne = headKeyOne; |
| long prevKeyTwo = headKeyTwo; |
| int currentCount = -1; |
| int newCount = -1; |
| String client = null; |
| |
| // Always printing this out, really useful when debugging. |
| LOG.info("Head: " + getStringFromKeys(headKeyOne, headKeyTwo)); |
| |
| do { |
| RowResult prev = nextNode(prevKeyOne, prevKeyTwo); |
| if (prev == null) { |
| context.getCounter(Counts.BROKEN_LINKS).increment(1); |
| LOG.warn(getStringFromKeys(prevKeyOne, prevKeyTwo) + " doesn't exist"); |
| break; |
| } |
| |
| // It's possible those columns are null, let's not break trying to read them. |
| if (prev.isNull(0) || prev.isNull(1)) { |
| context.getCounter(Counts.BROKEN_LINKS).increment(1); |
| LOG.warn(getStringFromKeys(prevKeyOne, prevKeyTwo) + " isn't referencing anywhere"); |
| break; |
| } |
| |
| int prevCount = prev.getInt(2); |
| String prevClient = prev.getString(3); |
| if (currentCount == -1) { |
| // First time we loop we discover what the count was and set the new one. |
| currentCount = prevCount; |
| newCount = currentCount + 1; |
| client = prevClient; |
| } |
| |
| if (prevCount != currentCount) { |
| context.getCounter(Counts.BAD_UPDATE_COUNTS).increment(1); |
| LOG.warn(getStringFromKeys(prevKeyOne, prevKeyTwo) + " has a wrong updateCount, " + |
| prevCount + " instead of " + currentCount); |
| // Game over, there's corruption. |
| break; |
| } |
| |
| if (!prevClient.equals(client)) { |
| context.getCounter(Counts.BROKEN_LINKS).increment(1); |
| LOG.warn(getStringFromKeys(prevKeyOne, prevKeyTwo) + " has the wrong client, " + |
| "bad reference? Bad client= " + prevClient); |
| break; |
| } |
| |
| updateRow(prevKeyOne, prevKeyTwo, newCount); |
| context.getCounter(Counts.UPDATED_NODES).increment(1); |
| if (prevKeyOne % 10 == 0) { |
| context.progress(); |
| } |
| prevKeyOne = prev.getLong(0); |
| prevKeyTwo = prev.getLong(1); |
| } while (headKeyOne != prevKeyOne && headKeyTwo != prevKeyTwo); |
| |
| updateStatCounters(context, newCount); |
| context.getCounter(Counts.UPDATED_LINKS).increment(1); |
| } |
| |
| /** |
| * Finds the next node in the linked list. |
| */ |
| private RowResult nextNode(long prevKeyOne, long prevKeyTwo) throws KuduException { |
| KuduScanner.KuduScannerBuilder builder = client.newScannerBuilder(table) |
| .setProjectedColumnNames(SCAN_COLUMN_NAMES); |
| configureScannerForRandomRead(builder, table, prevKeyOne, prevKeyTwo); |
| return getOneRowResult(builder.build()); |
| } |
| |
| private void updateRow(long keyOne, long keyTwo, int newCount) throws IOException { |
| Update update = table.newUpdate(); |
| PartialRow row = update.getRow(); |
| row.addLong(COLUMN_KEY_ONE, keyOne); |
| row.addLong(COLUMN_KEY_TWO, keyTwo); |
| row.addInt(COLUMN_UPDATE_COUNT, newCount); |
| session.apply(update); |
| } |
| |
| /** |
| * We keep some statistics about the linked list we update so that we can get a feel of |
| * what's being updated. |
| */ |
| private void updateStatCounters(Mapper.Context context, int newCount) { |
| switch (newCount) { |
| case -1: |
| case 0: |
| // TODO We didn't event get the first node? |
| break; |
| case 1: |
| context.getCounter(Counts.FIRST_UPDATE).increment(1); |
| break; |
| case 2: |
| context.getCounter(Counts.SECOND_UPDATE).increment(1); |
| break; |
| case 3: |
| context.getCounter(Counts.THIRD_UPDATE).increment(1); |
| break; |
| case 4: |
| context.getCounter(Counts.FOURTH_UPDATE).increment(1); |
| break; |
| default: |
| context.getCounter(Counts.MORE_THAN_FOUR_UPDATES).increment(1); |
| break; |
| } |
| } |
| |
| @Override |
| protected void cleanup(Context context) throws KuduException { |
| session.close(); |
| client.shutdown(); |
| } |
| } |
| |
| public int run(long maxLinkUpdatesPerMapper) throws Exception { |
| LOG.info("Running Updater with maxLinkUpdatesPerMapper=" + maxLinkUpdatesPerMapper); |
| |
| Job job = new Job(getConf()); |
| |
| job.setJobName("Link Updater"); |
| job.setNumReduceTasks(0); |
| job.setJarByClass(getClass()); |
| |
| Joiner columnsToQuery = Joiner.on(","); |
| |
| new KuduTableMapReduceUtil.TableInputFormatConfiguratorWithCommandLineParser( |
| job, getHeadsTable(getConf()), |
| columnsToQuery.join(COLUMN_KEY_ONE, COLUMN_KEY_TWO)) |
| .configure(); |
| |
| job.setMapperClass(UpdaterMapper.class); |
| job.setMapOutputKeyClass(BytesWritable.class); |
| job.setMapOutputValueClass(BytesWritable.class); |
| job.getConfiguration().setBoolean("mapred.map.tasks.speculative.execution", false); |
| // If something fails we want to exit ASAP. |
| job.getConfiguration().setInt("mapreduce.map.maxattempts", 1); |
| // Lack of YARN-445 means we can't auto-jstack on timeout, so disabling the timeout gives |
| // us a chance to do it manually. |
| job.getConfiguration().setInt("mapreduce.task.timeout", 0); |
| job.getConfiguration().setLong(MAX_LINK_UPDATES_PER_MAPPER, maxLinkUpdatesPerMapper); |
| |
| job.setOutputKeyClass(NullWritable.class); |
| job.setOutputValueClass(NullWritable.class); |
| job.setOutputFormatClass(NullOutputFormat.class); |
| |
| KuduTableMapReduceUtil.addDependencyJars(job); |
| |
| boolean success = job.waitForCompletion(true); |
| |
| Counters counters = job.getCounters(); |
| |
| if (success) { |
| // Let's not continue looping if we have broken linked lists. |
| Counter brokenLinks = counters.findCounter(Counts.BROKEN_LINKS); |
| Counter badUpdates = counters.findCounter(Counts.BAD_UPDATE_COUNTS); |
| if (brokenLinks.getValue() > 0 || badUpdates.getValue() > 0) { |
| LOG.error("Corruption was detected, see the job's counters. Ending the update loop."); |
| success = false; |
| } |
| } |
| return success ? 0 : 1; |
| } |
| |
| @Override |
| public int run(String[] args) throws Exception { |
| if (args.length < 2) { |
| System.err.println("Usage: Update <num iterations> <max link updates per mapper>"); |
| System.err.println(" where <num iterations> will be 'infinite' if passed a negative value" + |
| " or zero"); |
| return 1; |
| } |
| LOG.info("Running Loop with args:" + Arrays.deepToString(args)); |
| |
| int numIterations = Integer.parseInt(args[0]); |
| long maxUpdates = Long.parseLong(args[1]); |
| |
| if (numIterations <= 0) { |
| numIterations = Integer.MAX_VALUE; |
| } |
| |
| if (maxUpdates < 1) { |
| maxUpdates = 1; |
| } |
| |
| for (int i = 0; i < numIterations; i++) { |
| LOG.info("Starting iteration = " + i); |
| int ret = run(maxUpdates); |
| if (ret != 0) { |
| LOG.error("Can't continue updating, last run failed."); |
| return ret; |
| } |
| } |
| |
| return 0; |
| } |
| } |
| |
| /** |
| * A stand alone program that deletes a single node. |
| * TODO |
| */ |
| /*private static class Delete extends Configured implements Tool { |
| @Override |
| public int run(String[] args) throws Exception { |
| if (args.length != 1) { |
| System.out.println("Usage : " + Delete.class.getSimpleName() + " <node to delete>"); |
| return 0; |
| } |
| byte[] val = Bytes.toBytesBinary(args[0]); |
| |
| org.apache.hadoop.hbase.client.Delete delete |
| = new org.apache.hadoop.hbase.client.Delete(val); |
| |
| HTable table = new HTable(getConf(), getTableName(getConf())); |
| |
| table.delete(delete); |
| table.flushCommits(); |
| table.close(); |
| |
| System.out.println("Delete successful"); |
| return 0; |
| } |
| }*/ |
| |
| /** |
| * A stand alone program that follows a linked list created by {@link Generator} |
| * and prints timing info. |
| * |
| */ |
| private static class Walker extends Configured implements Tool { |
| |
| private KuduClient client; |
| private KuduTable table; |
| |
| @Override |
| public int run(String[] args) throws IOException { |
| if (args.length < 1) { |
| System.err.println("Usage: Walker <start key> [<num nodes>]"); |
| System.err.println(" where <num nodes> defaults to 100 nodes that will be printed out"); |
| return 1; |
| } |
| int maxNumNodes = 100; |
| if (args.length == 2) { |
| maxNumNodes = Integer.parseInt(args[1]); |
| } |
| System.out.println("Running Walker with args:" + Arrays.deepToString(args)); |
| |
| String[] keys = args[0].split(","); |
| if (keys.length != 2) { |
| System.err.println("The row key must be formatted like key1,key2"); |
| return 1; |
| } |
| |
| long keyOne = Long.parseLong(keys[0]); |
| long keyTwo = Long.parseLong(keys[1]); |
| |
| System.out.println("Walking with " + getStringFromKeys(keyOne, keyTwo)); |
| |
| walk(keyOne, keyTwo, maxNumNodes); |
| return 0; |
| } |
| |
| private void walk(long headKeyOne, long headKeyTwo, int maxNumNodes) throws KuduException { |
| CommandLineParser parser = new CommandLineParser(getConf()); |
| client = parser.getClient(); |
| table = client.openTable(getTableName(getConf())); |
| |
| long prevKeyOne = headKeyOne; |
| long prevKeyTwo = headKeyTwo; |
| CINode node = new CINode(); |
| int nodesCount = 0; |
| |
| do { |
| RowResult rr = nextNode(prevKeyOne, prevKeyTwo); |
| if (rr == null) { |
| System.err.println(getStringFromKeys(prevKeyOne, prevKeyTwo) + " doesn't exist!"); |
| break; |
| } |
| getCINode(rr, node); |
| printCINodeString(node); |
| if (rr.isNull(2) || rr.isNull(3)) { |
| System.err.println("Last node didn't have a reference, breaking"); |
| break; |
| } |
| prevKeyOne = rr.getLong(2); |
| prevKeyTwo = rr.getLong(3); |
| nodesCount++; |
| } while ((headKeyOne != prevKeyOne && headKeyTwo != prevKeyTwo) && (nodesCount < |
| maxNumNodes)); |
| } |
| |
| private RowResult nextNode(long keyOne, long keyTwo) throws KuduException { |
| KuduScanner.KuduScannerBuilder builder = client.newScannerBuilder(table); |
| configureScannerForRandomRead(builder, table, keyOne, keyTwo); |
| |
| return getOneRowResult(builder.build()); |
| } |
| } |
| |
| private static void configureScannerForRandomRead(AbstractKuduScannerBuilder builder, |
| KuduTable table, |
| long keyOne, |
| long keyTwo) { |
| PartialRow lowerBound = table.getSchema().newPartialRow(); |
| lowerBound.addLong(0, keyOne); |
| lowerBound.addLong(1, keyTwo); |
| builder.lowerBound(lowerBound); |
| |
| PartialRow upperBound = table.getSchema().newPartialRow(); |
| // Adding 1 since we want a single row, and the upper bound is exclusive. |
| upperBound.addLong(0, keyOne + 1); |
| upperBound.addLong(1, keyTwo + 1); |
| builder.exclusiveUpperBound(upperBound); |
| } |
| |
| private static String getTableName(Configuration conf) { |
| return conf.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME); |
| } |
| |
| private static String getHeadsTable(Configuration conf) { |
| return conf.get(HEADS_TABLE_NAME_KEY, DEFAULT_HEADS_TABLE_NAME); |
| } |
| |
| private static CINode getCINode(RowResult result, CINode node) { |
| |
| node.key = getStringFromKeys(result.getLong(0), result.getLong(1)); |
| if (result.isNull(2) || result.isNull(3)) { |
| node.prev = "NO_REFERENCE"; |
| } else { |
| node.prev = getStringFromKeys(result.getLong(2), result.getLong(3)); |
| } |
| node.rowId = result.getInt(4); |
| node.client = result.getString(5); |
| node.updateCount = result.getInt(6); |
| return node; |
| } |
| |
| private static void printCINodeString(CINode node) { |
| System.out.printf("%s:%s:%012d:%s:%s\n", node.key, node.prev, node.rowId, node.client, |
| node.updateCount); |
| } |
| |
| private static String getStringFromKeys(long key1, long key2) { |
| return new StringBuilder().append(key1).append(",").append(key2).toString(); |
| } |
| |
| private static RowResult getOneRowResult(KuduScanner scanner) throws KuduException { |
| RowResultIterator rowResults; |
| rowResults = scanner.nextRows(); |
| if (rowResults.getNumRows() == 0) { |
| return null; |
| } |
| if (rowResults.getNumRows() > 1) { |
| throw new RuntimeException("Received too many rows from scanner " + scanner); |
| } |
| return rowResults.next(); |
| } |
| |
| private void usage() { |
| System.err.println("Usage: " + this.getClass().getSimpleName() + " COMMAND [COMMAND options]"); |
| System.err.println(" where COMMAND is one of:"); |
| System.err.println(""); |
| System.err.println(" Generator A map only job that generates data."); |
| System.err.println(" Verify A map reduce job that looks for holes"); |
| System.err.println(" Look at the counts after running"); |
| System.err.println(" REFERENCED and UNREFERENCED are ok"); |
| System.err.println(" any UNDEFINED counts are bad. Do not"); |
| System.err.println(" run at the same time as the Generator."); |
| System.err.println(" Print A standalone program that prints nodes"); |
| System.err.println(" in the linked list."); |
| System.err.println(" Loop A program to Loop through Generator and"); |
| System.err.println(" Verify steps"); |
| System.err.println(" Update A program to updade the nodes"); |
| /* System.err.println(" Delete A standalone program that deletes a"); |
| System.err.println(" single node.");*/ |
| System.err.println(" Walker A standalong program that starts "); |
| System.err.println(" following a linked list"); |
| System.err.println("\t "); |
| System.err.flush(); |
| } |
| |
| protected void processOptions(String[] args) { |
| //get the class, run with the conf |
| if (args.length < 1) { |
| usage(); |
| throw new RuntimeException("Incorrect Number of args."); |
| } |
| toRun = args[0]; |
| otherArgs = Arrays.copyOfRange(args, 1, args.length); |
| } |
| |
| @Override |
| public int run(String[] args) throws Exception { |
| Tool tool; |
| processOptions(args); |
| if (toRun.equals("Generator")) { |
| tool = new Generator(); |
| } else if (toRun.equals("Verify")) { |
| tool = new Verify(); |
| } else if (toRun.equals("Loop")) { |
| Loop loop = new Loop(); |
| loop.it = this; |
| tool = loop; |
| |
| } else if (toRun.equals("Print")) { |
| tool = new Print(); |
| } else if (toRun.equals("Update")) { |
| tool = new Updater(); |
| } else if (toRun.equals("Walker")) { |
| tool = new Walker(); |
| } else { |
| usage(); |
| throw new RuntimeException("Unknown arg"); |
| } |
| |
| return ToolRunner.run(getConf(), tool, otherArgs); |
| } |
| |
| private static void setJobConf(Job job, int numMappers, long numNodes, |
| Integer width, Integer wrapMultiplier) { |
| job.getConfiguration().setInt(GENERATOR_NUM_MAPPERS_KEY, numMappers); |
| job.getConfiguration().setLong(GENERATOR_NUM_ROWS_PER_MAP_KEY, numNodes); |
| if (width != null) { |
| job.getConfiguration().setInt(GENERATOR_WIDTH_KEY, width); |
| } |
| if (wrapMultiplier != null) { |
| job.getConfiguration().setInt(GENERATOR_WRAP_KEY, wrapMultiplier); |
| } |
| } |
| |
| public static void main(String[] args) throws Exception { |
| int ret = ToolRunner.run(new IntegrationTestBigLinkedList(), args); |
| System.exit(ret); |
| } |
| } |
