使用MapReduce实现专利文件的分析

这个是美国专利文件的下载的地址
http://www.nber.org/patents/

escription	Documentation	Data -- Pkzipped
SAS .tpt	ASCII CSV
Overview	overview.txt	--
Pairwise citations data	Cite75_99.txt	Cite75_99.zip -- (68 Mb)	acite75_99.zip -- (82 Mb)
Patent data, including constructed variables	pat63_99.txt	pat63_99.zip -- (90Mb)	apat63_99.zip -- (56Mb)
Assignee names	coname.txt	coname.zip -- (2Mb)	aconame.zip -- (2Mb)
Contains the match to CUSIP numbers	match.txt	match.zip -- (130Kb)	amatch.zip -- (98Kb)
Individual inventor records	inventor.txt	inventor.zip -- (98Mb)	ainventor.zip -- (82Mb)
Class codes with corresponding class names	classes.txt	--
Country codes with corresponding country names	countries.txt
Class, technological category, and technological subcategory crosswalk	class_match.txt
Technological category and subcategory labels	subcategory.txt	--	subcategory.csv
SAS program to convert .tpt files to native SAS format	--	read_tpt.sas	--
U.S. Patent Classification (USPC) System and the Standard Industrial Code (SIC) System

下载该网站的下面这二个文件作为分析对象

Cite75_99.zip -- (68 Mb)	acite75_99.zip -- (82 Mb)
Patent data, including constructed variables	pat63_99.txt	pat63_99.zip -- (90Mb)

第一MapReduce的要求就是

表示出某个专利，被哪些专利引用

package PatentStatistics;

import java.io.IOException;

/**

* 构建专利引用列表，输入专利引用关系的关系对(patentNo1,patentNo2) 文件，输出每个专利号的所引用的文件，以逗号相隔。

*

*/

public class PatentCitation {

public static class PatentCitationMapper extends

Mapper<LongWritable, Text, Text, Text> {

/**

* 输入键位行偏移，值为“专利号1,专利号2”

*/

@Override

public void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

String[] citation = value.toString().split(",");

context.write(new Text(citation[1]), new Text(citation[0]));

}

}

public static class PatentCitationReducer extends

Reducer<Text, Text, Text, Text> {

public void reduce(Text key, Iterable<Text> values, Context context)

throws IOException, InterruptedException {

StringBuilder csv = new StringBuilder("");

for (Text val : values) {

if (csv.length() > 0) {

csv.append(",");

}

csv.append(val.toString());

}

context.write(key, new Text(csv.toString()));

}

}

public static void main(String[] args) throws IOException,

InterruptedException, ClassNotFoundException {

Job patentCitationJob = new Job();

patentCitationJob.setJobName("patentCitationJob");

patentCitationJob.setJarByClass(PatentCitation.class);

patentCitationJob.setMapperClass(PatentCitationMapper.class);

patentCitationJob.setMapOutputKeyClass(Text.class);

patentCitationJob.setMapOutputValueClass(Text.class);

patentCitationJob.setReducerClass(PatentCitationReducer.class);

patentCitationJob.setOutputKeyClass(Text.class);

patentCitationJob.setOutputValueClass(Text.class);

patentCitationJob.setInputFormatClass(TextInputFormat.class);

patentCitationJob.setOutputFormatClass(TextOutputFormat.class);

FileInputFormat.addInputPath(patentCitationJob, new Path(args[0]));

FileOutputFormat.setOutputPath(patentCitationJob, new Path(args[1]));

patentCitationJob.waitForCompletion(true);

System.out.println("finished!");

}

}

2、专利被引用的次数的计算

package PatentStatistics;

import java.io.IOException;

/**

* 专利被引用次数统计

*/

public class CitationCount {

public static class PatentCitationMapper extends

Mapper<LongWritable, Text, Text, IntWritable> {

private IntWritable one = new IntWritable(1);

public void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

// 输入key: 行偏移值；value: “citing专利号, cited专利号” 数据对

String[] citation = value.toString().split(",");

// 输出key: cited 专利号；value: 1

context.write(new Text(citation[1]), one);

}

}

public static class ReduceClass extends

Reducer<Text, IntWritable, Text, IntWritable> {

public void reduce(Text key, Iterable<IntWritable> values,

Context context) throws IOException, InterruptedException {

int count = 0;

for (IntWritable val : values) {

count += val.get();

}

// 输出key: 被引专利号；value: 被引次数

context.write(key, new IntWritable(count));

}

}

public static void main(String[] args) throws IOException,

InterruptedException, ClassNotFoundException {

Job citationCountJob = new Job();

citationCountJob.setJobName("citationCountJob");

citationCountJob.setJarByClass(CitationCount.class);

citationCountJob.setMapperClass(PatentCitationMapper.class);

citationCountJob.setMapOutputKeyClass(Text.class);

citationCountJob.setMapOutputValueClass(IntWritable.class);

citationCountJob.setReducerClass(ReduceClass.class);

citationCountJob.setOutputKeyClass(Text.class);

citationCountJob.setOutputValueClass(IntWritable.class);

citationCountJob.setInputFormatClass(TextInputFormat.class);

citationCountJob.setOutputFormatClass(TextOutputFormat.class);

FileInputFormat.addInputPath(citationCountJob, new Path(args[0]));

FileOutputFormat.setOutputPath(citationCountJob, new Path(args[1]));

citationCountJob.waitForCompletion(true);

System.out.println("finished!");

}

}

3专利被引用次数的直方图，key就是被引用的数，value就是出现该数的个数

package PatentStatistics;

import java.io.IOException;

/**

* 专利被引用次数分布统计，输入文件为专利引用次数统计的输出结果 扫描文件忽略专利号，仅仅考虑被引用的次数，统计每一个次数分别 有多少次出现。

*

*/

public class CitationCountDistribution {

public static class MapClass extends

Mapper<Object, Text, IntWritable, IntWritable> {

private IntWritable one = new IntWritable(1);

@Override

protected void map(Object key, Text value, Context context)

throws IOException, InterruptedException {

IntWritable citationCount = new IntWritable(Integer.parseInt(value

.toString().split(" ")[1]));

context.write(citationCount, one);

}

}

public static class ReduceClass extends

Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {

public void reduce(IntWritable key, Iterable<IntWritable> values,

Context context) throws IOException, InterruptedException {

int count = 0;

for (IntWritable val : values) {

count += val.get();

}

// 输出key: 被引次数；value: 总出现次数

context.write(key, new IntWritable(count));

}

}

public static void main(String[] args) throws IOException,

InterruptedException, ClassNotFoundException {

Job citationCountDistributionJob = new Job();

citationCountDistributionJob.setJobName("citationCountDistributionJob");

citationCountDistributionJob

.setJarByClass(CitationCountDistribution.class);

citationCountDistributionJob.setMapperClass(MapClass.class);

citationCountDistributionJob.setMapOutputKeyClass(IntWritable.class);

citationCountDistributionJob.setMapOutputValueClass(IntWritable.class);

citationCountDistributionJob.setReducerClass(ReduceClass.class);

citationCountDistributionJob.setOutputKeyClass(IntWritable.class);

citationCountDistributionJob.setOutputValueClass(IntWritable.class);

citationCountDistributionJob.setInputFormatClass(TextInputFormat.class);

citationCountDistributionJob

.setOutputFormatClass(TextOutputFormat.class);

FileInputFormat.addInputPath(citationCountDistributionJob, new Path(

args[0]));

FileOutputFormat.setOutputPath(citationCountDistributionJob, new Path(

args[1]));

citationCountDistributionJob.waitForCompletion(true);

System.out.println("finished!");

}

}