{"type":"doc","content":[{"type":"panel","attrs":{"panelType":"warning"},"content":[{"type":"paragraph","content":[{"text":"Important:","type":"text","marks":[{"type":"strong"}]},{"text":" Apache Hadoop MapReduce is not supported in CDAP versions 6.7.0 and later. It will be removed in a future release.","type":"text"}]}]},{"type":"paragraph","content":[{"text":"A ","type":"text"},{"text":"MapReduce program","type":"text","marks":[{"type":"em"}]},{"text":" is used to process data in batch. MapReduce can be written as in a conventional Hadoop system. Additionally, CDAP datasets can be accessed from MapReduce as both input and output.","type":"text"}]},{"type":"paragraph","content":[{"text":"To process data using MapReduce, either specify ","type":"text"},{"text":"addMapReduce()","type":"text","marks":[{"type":"code"}]},{"text":" in your application specification:","type":"text"}]},{"type":"codeBlock","content":[{"text":"public void configure() {\n ...\n addMapReduce(new WordCountJob());","type":"text"}]},{"type":"paragraph","content":[{"text":"or specify ","type":"text"},{"text":"addWorkflow()","type":"text","marks":[{"type":"code"}]},{"text":" in your application and specify your MapReduce in the workflow definition:","type":"text"}]},{"type":"codeBlock","content":[{"text":"public void configure() {\n ...\n // Run a MapReduce on the acquired data using a workflow\n addWorkflow(new PurchaseHistoryWorkflow());","type":"text"}]},{"type":"paragraph","content":[{"text":"It is recommended to implement the ","type":"text"},{"text":"MapReduce","type":"text","marks":[{"type":"code"}]},{"text":" interface by extending the ","type":"text"},{"text":"AbstractMapReduce","type":"text","marks":[{"type":"code"}]},{"text":" class, which allows the overriding of these three methods:","type":"text"}]},{"type":"bulletList","content":[{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"configure()","type":"text","marks":[{"type":"code"}]}]}]},{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"initialize()","type":"text","marks":[{"type":"code"}]}]}]},{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"destroy()","type":"text","marks":[{"type":"code"}]}]}]}]},{"type":"codeBlock","content":[{"text":"public class PurchaseHistoryBuilder extends AbstractMapReduce {\n @Override\n public void configure() {\n setName(\"Purchase History Builder MapReduce\");\n setDescription(\"Builds a purchase history for each customer\");\n }","type":"text"}]},{"type":"paragraph","content":[{"text":"The configure method is similar to the one found in applications. It defines the name and description of the MapReduce program. You can also specify resources (memory and virtual cores) used by the mappers and reducers.","type":"text"}]},{"type":"paragraph","content":[{"text":"The ","type":"text"},{"text":"initialize()","type":"text","marks":[{"type":"code"}]},{"text":" method is invoked at runtime, before the MapReduce is executed. Through the ","type":"text"},{"text":"getContext()","type":"text","marks":[{"type":"code"}]},{"text":" method you can obtain an instance of the ","type":"text"},{"text":"MapReduceContext","type":"text","marks":[{"type":"code"}]},{"text":". It allows you to specify datasets to be used as input or output. It also provides you access to the actual Hadoop job configuration, as though you were running the MapReduce directly on Hadoop. For example, you can specify the input and output datasets, the mapper and reducer classes as well as the intermediate data format:","type":"text"}]},{"type":"codeBlock","content":[{"text":"@Override\npublic void initialize() throws Exception {\n MapReduceContext context = getContext();\n context.addInput(Input.ofDataset(\"purchases\"));\n context.addOutput(Output.ofDataset(\"history\"));\n\n Job job = context.getHadoopJob();\n job.setMapperClass(PurchaseMapper.class);\n job.setMapOutputKeyClass(Text.class);\n job.setMapOutputValueClass(Text.class);\n job.setReducerClass(PerUserReducer.class);\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"The ","type":"text"},{"text":"destroy()","type":"text","marks":[{"type":"code"}]},{"text":" method is invoked after the MapReduce has finished. You could perform cleanup or send a notification of job completion, if that was required. Through the ","type":"text"},{"text":"MapReduceContext","type":"text","marks":[{"type":"code"}]},{"text":", you can determine whether the MapReduce succeeded:","type":"text"}]},{"type":"codeBlock","content":[{"text":"@Override\npublic void destroy() {\n ProgramState state = getContext().getState();\n // do something depending on the state\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"Because many MapReduce programs do not need this method, the ","type":"text"},{"text":"AbstractMapReduce","type":"text","marks":[{"type":"code"}]},{"text":" class provides a default implementation that does nothing.","type":"text"}]},{"type":"paragraph","content":[{"text":"CDAP ","type":"text"},{"text":"Mapper","type":"text","marks":[{"type":"code"}]},{"text":" and ","type":"text"},{"text":"Reducer","type":"text","marks":[{"type":"code"}]},{"text":" implement ","type":"text"},{"text":"the standard Hadoop APIs","type":"text","marks":[{"type":"link","attrs":{"href":"http://hadoop.apache.org/docs/r2.3.0/api/org/apache/hadoop/mapreduce/package-summary.html"}}]},{"text":" (note that CDAP only supports the \"new\" API in ","type":"text"},{"text":"org.apache.hadoop.mapreduce","type":"text","marks":[{"type":"code"}]},{"text":", and not the \"old\" API in ","type":"text"},{"text":"org.apache.hadoop.mapred","type":"text","marks":[{"type":"code"}]},{"text":"):","type":"text"}]},{"type":"codeBlock","content":[{"text":"public static class PurchaseMapper\n extends Mapper {\n\n private Metrics mapMetrics;\n\n @Override\n public void map(byte[] key, Purchase purchase, Context context)\n throws IOException, InterruptedException {\n String user = purchase.getCustomer();\n if (purchase.getPrice() > 100000) {\n mapMetrics.count(\"purchases.large\", 1);\n }\n context.write(new Text(user), new Text(new Gson().toJson(purchase)));\n }\n}\n\npublic static class PerUserReducer\n extends Reducer {\n\n @UseDataSet(\"frequentCustomers\")\n private KeyValueTable frequentCustomers;\n private Metrics reduceMetrics;\n\n public void reduce(Text customer, Iterable values, Context context)\n throws IOException, InterruptedException {\n PurchaseHistory purchases = new PurchaseHistory(customer.toString());\n int numPurchases = 0;\n for (Text val : values) {\n purchases.add(new Gson().fromJson(val.toString(), Purchase.class));\n numPurchases++;\n }\n if (numPurchases == 1) {\n reduceMetrics.count(\"customers.rare\", 1);\n } else if (numPurchases > 10) {\n reduceMetrics.count(\"customers.frequent\", 1);\n frequentCustomers.write(customer.toString(), String.valueOf(numPurchases));\n }\n context.write(customer.toString(), purchases);\n }\n}","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"MapReduce and Datasets","type":"text"}]},{"type":"paragraph","content":[{"text":"Both a CDAP ","type":"text"},{"text":"Mapper","type":"text","marks":[{"type":"code"}]},{"text":" and ","type":"text"},{"text":"Reducer","type":"text","marks":[{"type":"code"}]},{"text":" can directly read or write to a dataset, using one of the following options. (Note that the second and third options can be used for a ","type":"text"},{"text":"Partitioner","type":"text","marks":[{"type":"code"}]},{"text":" or ","type":"text"},{"text":"Comparator","type":"text","marks":[{"type":"code"}]},{"text":", if configured on the MapReduce job.)","type":"text"}]},{"type":"orderedList","attrs":{"order":1},"content":[{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"Inject the dataset into the mapper or reducer that uses it. This method is useful if the name of the dataset is constant or known at compile time. For example, to have access to a dataset named ","type":"text"},{"text":"catalog","type":"text","marks":[{"type":"em"}]},{"text":":","type":"text"}]},{"type":"codeBlock","content":[{"text":"public static class CatalogJoinMapper extends Mapper {\n\n @UseDataSet(\"catalog\")\n private ProductCatalog catalog;\n\n @Override\n public void map(byte[] key, Purchase purchase, Context context)\n throws IOException, InterruptedException {\n // join with catalog by product ID\n Product product = catalog.read(purchase.getProductId());\n ...\n }","type":"text"}]}]},{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"Acquire the dataset in the mapper's or reducer's ","type":"text"},{"text":"initialize()","type":"text","marks":[{"type":"code"}]},{"text":" method. As opposed to the previous method, this does not require the dataset name to be constant; it only needs to be known at the time the task starts (for example, through configuration). Note that this requires that the mapper or reducer class implements the ","type":"text"},{"text":"ProgramLifecycle","type":"text","marks":[{"type":"code"}]},{"text":" interface, which includes the two methods ","type":"text"},{"text":"initialize()","type":"text","marks":[{"type":"code"}]},{"text":" and ","type":"text"},{"text":"destroy()","type":"text","marks":[{"type":"code"}]},{"text":":","type":"text"}]},{"type":"codeBlock","content":[{"text":"public static class CatalogJoinMapper extends Mapper\n implements ProgramLifecycle {\n\n private ProductCatalog catalog;\n\n @Override\n public void initialize(MapReduceTaskContext mapReduceTaskContext) throws Exception {\n catalog = mapReduceTaskContext.getDataset(\n mapReduceTaskContext.getRuntimeArguments().get(\"catalog.table.name\"));\n }\n\n @Override\n public void destroy() {\n }\n\n @Override\n public void map(byte[] key, Purchase purchase, Context context)\n throws IOException, InterruptedException {\n // join with catalog by product ID\n Product product = catalog.read(purchase.getProductId());\n ...\n }","type":"text"}]}]},{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"Dynamically acquire the dataset in the mapper every time it is accessed. This is useful if the name of the dataset is not known at initialization time; for example, if it depends on the data passed to each ","type":"text"},{"text":"map()","type":"text","marks":[{"type":"code"}]},{"text":" call. In this case, you also implement the ","type":"text"},{"text":"ProgramLifecycle","type":"text","marks":[{"type":"code"}]},{"text":" interface, to save the ","type":"text"},{"text":"MapReduceTaskContext","type":"text","marks":[{"type":"code"}]},{"text":" for use in the ","type":"text"},{"text":"map()","type":"text","marks":[{"type":"code"}]},{"text":" method. For example:","type":"text"}]},{"type":"codeBlock","content":[{"text":"public static class CatalogJoinMapper extends Mapper\n implements ProgramLifecycle {\n\n private MapReduceTaskContext taskContext;\n\n @Override\n public void initialize(MapReduceTaskContext mapReduceTaskContext) throws Exception {\n taskContext = mapReduceTaskContext;\n }\n\n @Override\n public void destroy() {\n }\n\n @Override\n public void map(byte[] key, Purchase purchase, Context context)\n throws IOException, InterruptedException {\n // join with catalog by product ID\n String catalogName = determineCatalogName(purchase.getProductCategory());\n ProductCatalog catalog = taskContext.getDataset(catalogName);\n Product product = catalog.read(purchase.getProductId());\n ...\n }\n\n private String determineCatalogName(String productCategory) {\n ...\n }","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"See also the section on ","type":"text"},{"text":"Using Datasets in Programs","type":"text","marks":[{"type":"link","attrs":{"href":"https://cdap.atlassian.net/wiki/spaces/DOCS/pages/480313614/Datasets+overview"}}]},{"text":".","type":"text"}]},{"type":"paragraph","content":[{"text":"A MapReduce program can interact with a dataset by using it as an input or an output. The dataset needs to implement specific interfaces to support this, as described in the following sections.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"A Dataset as the Input Source of a MapReduce Program","type":"text"}]},{"type":"paragraph","content":[{"text":"When you run a MapReduce program, you can configure it to read its input from a dataset. The source dataset must implement the ","type":"text"},{"text":"BatchReadable","type":"text","marks":[{"type":"code"}]},{"text":" interface, which requires two methods:","type":"text"}]},{"type":"codeBlock","content":[{"text":"public interface BatchReadable {\n List getSplits();\n SplitReader createSplitReader(Split split);\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"These two methods complement each other: ","type":"text"},{"text":"getSplits()","type":"text","marks":[{"type":"code"}]},{"text":" must return all splits of the dataset that the MapReduce program will read; ","type":"text"},{"text":"createSplitReader()","type":"text","marks":[{"type":"code"}]},{"text":" is then called in every Mapper to read one of the splits. Note that the ","type":"text"},{"text":"KEY","type":"text","marks":[{"type":"code"}]},{"text":" and ","type":"text"},{"text":"VALUE","type":"text","marks":[{"type":"code"}]},{"text":" type parameters of the split reader must match the input key and value type parameters of the Mapper.","type":"text"}]},{"type":"paragraph","content":[{"text":"Because ","type":"text"},{"text":"getSplits()","type":"text","marks":[{"type":"code"}]},{"text":" has no arguments, it will typically create splits that cover the entire dataset. If you want to use a custom selection of the input data, define another method in your dataset with additional parameters and explicitly set the input in the ","type":"text"},{"text":"initialize()","type":"text","marks":[{"type":"code"}]},{"text":" method.","type":"text"}]},{"type":"paragraph","content":[{"text":"For example, the system dataset ","type":"text"},{"text":"KeyValueTable","type":"text","marks":[{"type":"code"}]},{"text":" implements ","type":"text"},{"text":"BatchReadable","type":"text","marks":[{"type":"code"}]},{"text":" with an extra method that allows specification of the number of splits and a range of keys:","type":"text"}]},{"type":"codeBlock","content":[{"text":"public class KeyValueTable extends AbstractDataset\n implements BatchReadable {\n ...\n public List getSplits(int numSplits, byte[] start, byte[] stop);\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"To read a range of keys and give a hint that you want 16 splits, write:","type":"text"}]},{"type":"codeBlock","content":[{"text":"@UseDataSet(\"myTable\")\nKeyValueTable kvTable;\n...\n@Override\npublic void initialize() throws Exception {\n MapReduceContext context = getContext();\n ...\n context.addInput(Input.ofDataset(\"myTable\", kvTable.getSplits(16, startKey, stopKey)));\n}","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"A Dataset as the Output Destination of a MapReduce Program","type":"text"}]},{"type":"paragraph","content":[{"text":"Just as you have the option to read input from a dataset, you have the option to write to a dataset as the output destination of a MapReduce program if that dataset implements the ","type":"text"},{"text":"BatchWritable","type":"text","marks":[{"type":"code"}]},{"text":" interface:","type":"text"}]},{"type":"codeBlock","content":[{"text":"public interface BatchWritable {\n void write(KEY key, VALUE value);\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"The ","type":"text"},{"text":"write()","type":"text","marks":[{"type":"code"}]},{"text":" method is used to redirect all writes performed by a Reducer to the dataset. Again, the ","type":"text"},{"text":"KEY","type":"text","marks":[{"type":"code"}]},{"text":" and ","type":"text"},{"text":"VALUE","type":"text","marks":[{"type":"code"}]},{"text":" type parameters must match the output key and value type parameters of the Reducer.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Multiple Output Destinations of a MapReduce Program","type":"text"}]},{"type":"paragraph","content":[{"text":"To write to multiple output datasets from a MapReduce program, begin by adding the datasets as outputs:","type":"text"}]},{"type":"codeBlock","content":[{"text":"public void initialize() throws Exception {\n ...\n getContext().addOutput(Output.ofDataset(\"productCounts\"));\n getContext().addOutput(Output.ofDataset(\"catalog\"));\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"Then, have the ","type":"text"},{"text":"Mapper","type":"text","marks":[{"type":"code"}]},{"text":" and/or ","type":"text"},{"text":"Reducer","type":"text","marks":[{"type":"code"}]},{"text":" implement ","type":"text"},{"text":"ProgramLifeCycle","type":"text","marks":[{"type":"code"}]},{"text":". This is to obtain access to the ","type":"text"},{"text":"MapReduceTaskContext","type":"text","marks":[{"type":"code"}]},{"text":" in their initialization methods and to be able to write using the write method of the ","type":"text"},{"text":"MapReduceTaskContext","type":"text","marks":[{"type":"code"}]},{"text":":","type":"text"}]},{"type":"codeBlock","content":[{"text":"public static class CustomMapper extends Mapper\n implements ProgramLifecycle> {\n\n private MapReduceTaskContext mapReduceTaskContext;\n\n @Override\n public void initialize(MapReduceTaskContext context) throws Exception {\n this.mapReduceTaskContext = context;\n }\n\n protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {\n // compute some condition\n ...\n if (someCondition) {\n mapReduceTaskContext.write(\"productCounts\", key, value);\n } else {\n mapReduceTaskContext.write(\"catalog\", key, value);\n }\n }\n\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"Note that the multiple output write method, ","type":"text"},{"text":"MapReduceTaskContext.write(String, KEY key, VALUE value)","type":"text","marks":[{"type":"code"}]},{"text":", can only be used if there are multiple outputs. Similarly, the single output write method, ","type":"text"},{"text":"MapReduceTaskContext.write(KEY key, VALUE value)","type":"text","marks":[{"type":"code"}]},{"text":", can only be used if there is a single output to the MapReduce program.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"MapReduce and Transactions","type":"text"}]},{"type":"paragraph","content":[{"text":"When you run a MapReduce that interacts with datasets, the system creates a long-running transaction. Similar to the transaction of a service handler, here are some rules to follow:","type":"text"}]},{"type":"bulletList","content":[{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"Reads can only see the writes of other transactions that were committed at the time the long-running transaction was started.","type":"text"}]}]},{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"All writes of the long-running transaction are committed atomically, and only become visible to other transactions after they are committed.","type":"text"}]}]},{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"Tasks of a MapReduce program may not see each other's writes because of buffering of datasets.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"However, there is a key difference: long-running transactions do not participate in conflict detection. If another transaction overlaps with the long-running transaction and writes to the same row, it will not cause a conflict but simply overwrite it.","type":"text"}]},{"type":"paragraph","content":[{"text":"It is not efficient to fail the long-running job based on a single conflict. Because of this, it is not recommended to write to the same dataset from both real-time and MapReduce programs. It is better to use different datasets, or at least ensure that the real-time processing writes to a disjoint set of columns.","type":"text"}]},{"type":"paragraph","content":[{"text":"It's important to note that the MapReduce framework will reattempt a task (Mapper or Reducer) if it fails. If the task is writing to a dataset, the reattempt of the task will most likely repeat the writes that have been already performed in the failed attempt. Therefore, it is highly advisable that all writes performed by MapReduce programs be idempotent.","type":"text"}]},{"type":"paragraph","content":[{"text":"If your Map Reduce Job performs speculative execution, a similar repetition of of writes can occur.","type":"text"}]},{"type":"paragraph","content":[{"text":"See ","type":"text"},{"text":"Transaction System","type":"text","marks":[{"type":"link","attrs":{"href":"https://cdap.atlassian.net/wiki/spaces/DOCS/pages/480346480/Transaction+System"}}]},{"text":" for additional information.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"MapReduce and Resources","type":"text"}]},{"type":"paragraph","content":[{"text":"Both the YARN container size and the number of virtual cores used in a MapReduce job can be specified as part of the configuration. They can also be set at runtime through the use of runtime arguments.","type":"text"}]},{"type":"paragraph","content":[{"text":"The Resources API, if called with two arguments, sets both the memory used in megabytes and the number of virtual cores used.","type":"text"}]}],"version":1}

Browser not supported