Grep example-MaxCompute(MaxCompute)-阿里云帮助中心

This example runs a Grep job using MaxCompute MapReduce. The job scans an input table for records matching a regex pattern, counts the occurrences of each match, and writes the sorted results to an output table.

How it works

mr_src (input) → RegexMapper → mr_grep_tmp (intermediate) → LongSumReducer + InverseMapper → mr_grep_out (output)

RegexMapper scans each field in the input table and emits <matched_string, 1> for every regex match.
LongSumReducer aggregates the counts by matched string and writes <matched_string, count> to the intermediate table.
InverseMapper swaps keys and values to <count, matched_string> so the final output is sorted by occurrence count.

Prerequisites

Before you begin, ensure that you have:

Completed the environment setup described in Getting started

Prepare resources

Place the mapreduce-examples.jar file in the bin\data\resources directory of your MaxCompute client installation.

Create the tables required by the job.

CREATE TABLE mr_src(key STRING, value STRING);
CREATE TABLE mr_grep_tmp (key STRING, cnt BIGINT);
CREATE TABLE mr_grep_out (key BIGINT, value STRING);

Add the JAR as a MaxCompute resource.

-- Omit -f if this is the first time you add the JAR.
add jar data\resources\mapreduce-examples.jar -f;

Use Tunnel to load the sample data into mr_src.
```
tunnel upload data.txt mr_src;
```
The data.txt file is in the bin directory of the MaxCompute client. After the upload, mr_src contains:
```
hello,odps
hello,world
```

Run the Grep job

Run the following command on the MaxCompute client, passing hello as the regex pattern:

jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.Grep mr_src mr_grep_tmp mr_grep_out hello;

Command arguments

Argument	Description	Example
`<inDir>`	Input table	`mr_src`
`<tmpDir>`	Intermediate table for per-string counts	`mr_grep_tmp`
`<outDir>`	Output table sorted by count	`mr_grep_out`
`<regex>`	Pattern to match against each field	`hello`
`[<group>]`	Optional: capture group index (default: `0`)	`1`

Verify the result

After the job completes, query the output table:

SELECT * FROM mr_grep_out;

Expected output:

+------------+------------+
| key        | value      |
+------------+------------+
| 2          | hello      |
+------------+------------+

The result shows that hello matched 2 times in mr_src. Because InverseMapper swaps count and word before the final sort, the output key column holds the count and value holds the matched string.

Sample code

Note

For Project Object Model (POM) dependency configuration, see the Precautions section in the Getting started guide.

package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import java.util.Iterator;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.TableInfo;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.Mapper;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.ReducerBase;
import com.aliyun.odps.mapred.RunningJob;
import com.aliyun.odps.mapred.TaskContext;
import com.aliyun.odps.mapred.conf.JobConf;
import com.aliyun.odps.mapred.utils.InputUtils;
import com.aliyun.odps.mapred.utils.OutputUtils;
import com.aliyun.odps.mapred.utils.SchemaUtils;
/**
     *
     * Extracts matching regexs from input files and counts them.
     *
     **/
public class Grep {
    /**
       * RegexMapper
       **/
    public static class RegexMapper extends MapperBase {
        private Pattern pattern;
        private int group;
        private Record word;
        private Record one;
        @Override
            public void setup(TaskContext context) throws IOException {
            JobConf job = (JobConf) context.getJobConf();
            pattern = Pattern.compile(job.get("mapred.mapper.regex"));
            group = job.getInt("mapred.mapper.regex.group", 0);
            word = context.createMapOutputKeyRecord();
            one = context.createMapOutputValueRecord();
            one.set(new Object[] { 1L });
        }
        @Override
            public void map(long recordNum, Record record, TaskContext context) throws IOException {
            for (int i = 0; i < record.getColumnCount(); ++i) {
                String text = record.get(i).toString();
                Matcher matcher = pattern.matcher(text);
                while (matcher.find()) {
                    word.set(new Object[] { matcher.group(group) });
                    context.write(word, one);
                }
            }
        }
    }
    /**
       * LongSumReducer
       **/
    public static class LongSumReducer extends ReducerBase {
        private Record result = null;
        @Override
            public void setup(TaskContext context) throws IOException {
            result = context.createOutputRecord();
        }
        @Override
            public void reduce(Record key, Iterator<Record> values, TaskContext context) throws IOException {
            long count = 0;
            while (values.hasNext()) {
                Record val = values.next();
                count += (Long) val.get(0);
            }
            result.set(0, key.get(0));
            result.set(1, count);
            context.write(result);
        }
    }
    /**
       * A {@link Mapper} that swaps keys and values.
       **/
    public static class InverseMapper extends MapperBase {
        private Record word;
        private Record count;
        @Override
            public void setup(TaskContext context) throws IOException {
            word = context.createMapOutputValueRecord();
            count = context.createMapOutputKeyRecord();
        }
        /**
         * The inverse function. Input keys and values are swapped.
         **/
        @Override
            public void map(long recordNum, Record record, TaskContext context) throws IOException {
            word.set(new Object[] { record.get(0).toString() });
            count.set(new Object[] { (Long) record.get(1) });
            context.write(count, word);
        }
    }
    /**
       * IdentityReducer
       **/
    public static class IdentityReducer extends ReducerBase {
        private Record result = null;
        @Override
            public void setup(TaskContext context) throws IOException {
            result = context.createOutputRecord();
        }
        /** Writes all keys and values directly to output. **/
        @Override
            public void reduce(Record key, Iterator<Record> values, TaskContext context) throws IOException {
            result.set(0, key.get(0));
            while (values.hasNext()) {
                Record val = values.next();
                result.set(1, val.get(0));
                context.write(result);
            }
        }
    }
    public static void main(String[] args) throws Exception {
        if (args.length < 4) {
            System.err.println("Grep <inDir> <tmpDir> <outDir> <regex> [<group>]");
            System.exit(2);
        }
        JobConf grepJob = new JobConf();
        grepJob.setMapperClass(RegexMapper.class);
        grepJob.setReducerClass(LongSumReducer.class);
        grepJob.setMapOutputKeySchema(SchemaUtils.fromString("word:string"));
        grepJob.setMapOutputValueSchema(SchemaUtils.fromString("count:bigint"));
        InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), grepJob);
        OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), grepJob);
        /**Set the regular expression for grepJob. */
        grepJob.set("mapred.mapper.regex", args[3]);
        if (args.length == 5) {
            grepJob.set("mapred.mapper.regex.group", args[4]);
        }
        @SuppressWarnings("unused")
            RunningJob rjGrep = JobClient.runJob(grepJob);
        /**Specify the output of grepJob as the input of sortJob. */
        JobConf sortJob = new JobConf();
        sortJob.setMapperClass(InverseMapper.class);
        sortJob.setReducerClass(IdentityReducer.class);
        sortJob.setMapOutputKeySchema(SchemaUtils.fromString("count:bigint"));
        sortJob.setMapOutputValueSchema(SchemaUtils.fromString("word:string"));
        InputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), sortJob);
        OutputUtils.addTable(TableInfo.builder().tableName(args[2]).build(), sortJob);
        sortJob.setNumReduceTasks(1); // write a single file
        sortJob.setOutputKeySortColumns(new String[] { "count" });
        @SuppressWarnings("unused")
            RunningJob rjSort = JobClient.runJob(sortJob);
    }
}

Code walk-through

The Grep job runs as two chained MapReduce passes.

Pass 1 — count matches

RegexMapper.setup() reads mapred.mapper.regex from JobConf and compiles it into a Pattern. For each record, map() iterates over every column, applies the pattern, and emits <matched_string, 1> for each match found. LongSumReducer.reduce() sums the values for each unique key and writes <matched_string, total_count> to mr_grep_tmp.

Pass 2 — sort by count

InverseMapper.map() reads each row from mr_grep_tmp and swaps the key and value, emitting <count, matched_string>. Because the output key is now the count (a BIGINT), setting setOutputKeySortColumns(new String[]{"count"}) with a single reducer produces a globally sorted output. IdentityReducer.reduce() writes all key-value pairs to mr_grep_out unchanged.