会python真的可以为所欲为0(回)
349天前
这里还有人吗1(回)
534天前这里还有人吗0(回)
534天前
每天面对着电脑屏幕，敲打键盘。我所面对的并不只是代码，而是一种生活方式。0(回)
759天前到处都是羊，不想上班0(回)
839天前
鸽子
0(回)
843天前package org.platform.modules.elasticsearch.clean.ly; |
import java.io.IOException; |
import java.util.HashSet; |
import java.util.Map; |
import org.apache.hadoop.conf.Configuration; |
import org.apache.hadoop.fs.Path; |
import org.apache.hadoop.io.LongWritable; |
import org.apache.hadoop.io.Text; |
import org.apache.hadoop.mapreduce.Job; |
import org.apache.hadoop.mapreduce.Mapper; |
import org.apache.hadoop.mapreduce.Reducer; |
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; |
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; |
import org.apache.hadoop.util.GenericOptionsParser; |
import org.platform.modules.elasticsearch.clean.util.CleanUtil; |
import com.google.gson.Gson; |
import com.google.gson.GsonBuilder; |
/** |
* 找出同时存在新浪账号和密码的记录，并保存 |
*/ |
public class FindSina { |
public static class Map1 extends Mapper<LongWritable, Text, Text, Text>{ |
|
public void setup(Context context) throws IOException, InterruptedException { |
super.setup(context); |
this.gson = new GsonBuilder().serializeSpecialFloatingPointValues() |
.setDateFormat("yyyy-MM-dd HH:mm:ss").create(); |
} |
|
public Gson gson = null; |
public String emailReg = "[\\w!#$%&'*+//=?^_`{|}~-]+(?:\\.[\\w!#$%&'*+//=?^_`{|}~-]+)*@sina\\.((com)|(cn))"; |
@SuppressWarnings("unchecked") |
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{ |
try { |
Map<String, Object> original = gson.fromJson(value.toString(), Map.class); |
Map<String, Object> map = CleanUtil.replaceSpace(original); |
if(original.containsKey("email") && original.containsKey("passWord")){ |
String email = (String)map.get("email"); |
String passWord = (String)map.get("passWord"); |
if(email.matches(emailReg) && !"".equals(passWord) && !"NA".equals(passWord) && !"null".equals(passWord)){ |
Text emailText = new Text(email); |
Text passWordText = new Text(passWord); |
context.write(emailText, passWordText); |
} |
} |
} catch (Exception e) { |
e.printStackTrace(); |
} |
} |
} |
public static class Reduce extends Reducer<Text, Text, Text, Text>{ |
|
public void reduce(Text key , Iterable<Text> value,Context context) throws IOException, InterruptedException{ |
StringBuffer password = new StringBuffer(); |
//考虑密码重复的记录 |
HashSet<String> set = new HashSet<String>(); |
for(Text str : value){ |
if(!set.contains(str.toString())){ |
if(password.length()==0){ |
password = new StringBuffer(str.toString()); |
}else{ |
password.append(","+str.toString()); |
} |
set.add(str.toString()); |
}else{ |
set.add(str.toString()); |
} |
} |
Text passwordText = new Text(password.toString()); |
context.write(key, passwordText); |
} |
} |
|
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{ |
Configuration configuration = new Configuration(); |
configuration.set("mapred.job.tracker", "192.168.0.10"); |
args = new String[]{"hdfs://192.168.0.10:9000/elasticsearch_original/email/mailbox/20/records-10-m-00008","hdfs://192.168.0.10:9000/user/test/out/records-10-m-00008"}; |
String[] inputArgs = new GenericOptionsParser(configuration, args).getRemainingArgs(); |
if (inputArgs.length != 2) { |
System.out.println("error, please input two path. input and output"); |
System.exit(2); |
} |
Job job = Job.getInstance(configuration, "FindSina"); |
job.setJarByClass(FindSina.class); |
|
job.setMapperClass(Map1.class); |
job.setReducerClass(Reduce.class); |
|
job.setOutputKeyClass(Text.class); |
job.setOutputValueClass(Text.class); |
|
FileInputFormat.addInputPath(job, new Path(inputArgs[0])); |
FileOutputFormat.setOutputPath(job, new Path(inputArgs[1])); |
System.exit(job.waitForCompletion(true) ? 0 : 1); |
} |
|
} |
