工作中需要根据不同的值来设置不同的输出目录，有两个点需要注意：
1 其中参数2的 namedOutput 必须设置
MultipleOutputs.addNamedOutput(job, a, TextOutputFormat.class,
NullWritable.class, Text.class);
2
mos.write(keyStr, NullWritable.get(), valText, keyStr + "/"); // 对应于 run方法中的 addNamedOutput里的参数2， 这里必须保持一致 否则报Named ouput 'xxx' not defined的错，
具体代码如下，针对 mos.write中的baseOutputPath的不同设置的值的hdfs输出目录见代码注释：
import com.alibaba.fastjson.JSON;
import com.mydb.bigdata.config.Config;
import com.mydb.bigdata.xetl.model.Constant;
import com.mydb.bigdata.xetl.utils.AutoActLogParseUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
* Created by pc on 2017/9/21.
./log/data_log_parse_to_hdfs/data_log_parse_to_hdfs.sh:66:hadoop jar ${com_jar_bigdata_common_mongo_batch_get} com.mydb.bigdata.xetl.mr.AutoActLogParseMr
/collect_data/userlog/20170902/*userlog*.log.gz ----> 源头数据 args[0]
/log_data/2018-04-03/ ----> 解析后的文件夹 args[1]
'stg_log_' ------> 解析后的埋点表名 args[2]
/include/xetl.properties ------> hive源数据库的连接信息 args[3]
*
* create external table stg_log_1900039(...) PARTITIONED BY (day STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\001' LINES TERMINATED BY '\n' STORED AS TEXTFILE LOCATION '/log_data/stg_log_1900039';
*
*/
public class AutoActLogParseMr extends Configured implements Tool {
private static Logger logger = LoggerFactory.getLogger(AutoActLogParseMr.class);
// public static String day="";
public static String acts = "";
/**
*
* @param args
* 1、传入参数
* 2、传出参数
* 3、业务参数：①stg_log_ ②stg_log_class_perform_
* 4、配置文件路径：如xetl.properties
*/
public static void main(String[] args) {
if (args.length < 2) {
System.out.println("args must more than 2.");
System.exit(0);
}
// day=(args[0].split("/"))[3];
Configuration conf = new Configuration();
FileSystem hdfs = null;
try {
int res = ToolRunner.run(conf, new AutoActLogParseMr(), args);
System.exit(res);
} catch (Exception e) {
logger.error("", e);
}
}
public int run(String[] params) throws Exception {
Configuration conf = getConf();
conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
Integer numReduceTasks = 3;
FileSystem hdfs = null;
try {
// 程序配置
// conf.set("fs.default.name", "hdfs://Galaxy");
//config.set("hadoop.job.ugi", "feng,111111");
//config.set("hadoop.tmp.dir", "/tmp/hadoop-fengClient");
//config.set("dfs.replication", "1");
//config.set("mapred.job.tracker", "master:9001");
// hdfs = FileSystem.get(new URI("hdfs://Galaxy"),
// conf, "bigdata");
Path path = new Path("/log_data/");
hdfs = path.getFileSystem(conf);
// logger.info("path 的值：" + path);
String flag=params[2]; // stg_log_
acts = getOutPutName(hdfs, path, conf,flag); // 1900039@1900038
conf.set("fs.allActs", acts);
} catch (Exception e) {
e.printStackTrace();
}
// acts = Hdfstools.readHDFSFile("/log_data/actId");
// logger.info("acts的值为" + acts);
//获取配置文件信息
Config propertiesConfig = new Config();
propertiesConfig.init(params[3]); // xetl.properties
String mysqlUrl = propertiesConfig.getValue("mysqlUrl");
String mysqlUser = propertiesConfig.getValue("mysqlUser");
String mysqlPassword = propertiesConfig.getValue("mysqlPassword");
String dbname = propertiesConfig.getValue("dbname");
conf.set("mysqlUser",mysqlUser);
conf.set("mysqlUrl",mysqlUrl);
conf.set("mysqlPassword",mysqlPassword);
conf.set("dbname",dbname);
Job job = Job.getInstance(conf);
job.setJarByClass(AutoActLogParseMr.class);
job.setMapperClass(AutoActLogParseMr.AutoActLogParseMaper.class);
job.setReducerClass(AutoActLogParseMr.AutoActLogParseReducer.class);
//将第一个路径参数作为输入参数
FileInputFormat.setInputPaths(job, new Path(params[0])); // /collect_data/userlog/20170902/*userlog*.log.gz
//将第二个参数作为输出参数
FileOutputFormat.setOutputPath(job, new Path(params[1])); // /log_data/2018-04-03/
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setNumReduceTasks(numReduceTasks);
String dirName[] = acts.split(Constant.MARK_AITE); // 1900039@1900038 这些要解析的埋点变成数组
for (String a : dirName) { // hdfs多路径输出文件 ， 其中参数2的 namedOutput 必须设置
MultipleOutputs.addNamedOutput(job, a, TextOutputFormat.class,
NullWritable.class, Text.class);
}
logger.info("---excuter---");
return job.waitForCompletion(true) ? 0 : 1;
}
public static class AutoActLogParseMaper extends Mapper<LongWritable, Text, Text, Text> {
public static Map<String, List> actMap = new HashMap();
@Override
protected void setup(Context context) throws IOException, InterruptedException {
try {
String mysqlUser = context.getConfiguration().get("mysqlUser");
String mysqlUrl = context.getConfiguration().get("mysqlUrl");
String mysqlPassword = context.getConfiguration().get("mysqlPassword");
String dbname = context.getConfiguration().get("dbname");
String string = context.getConfiguration().get("fs.allActs"); // 1900039@1900038 要解析的actid
actMap = AutoActLogParseUtil.getHiveStaticConf(string,mysqlUrl,mysqlUser,mysqlPassword,dbname); // actids,mysql链接信息 得到 <actid, actid所在的表的所有列>
} catch (SQLException e) {
e.printStackTrace();
}
super.setup(context);
}
@Override
protected void cleanup(Context context)
throws IOException, InterruptedException {
}
protected void map(LongWritable key, Text val, Context context) {
try {
String jsonStr = val.toString();
if (StringUtils.isEmpty(jsonStr)) {
return;
}
/**
* 要解析的一行内容：
* 2017-09-02 00:00:17.017 172.16.0.8 openApi -- {"actId":"1712012","classId":"f17ee6fd1fcb4f1fb3e1688f5cd364bd","className":"九年级(1)班","countyId":"445381","countyName":"罗定市","gradeId":"9","originCode":"4","role":"STUDENT","schoolId"
:"18738","schoolName":"罗定第二中学","time":1504281617255,"userIcon":"d355db9797f854e1805e653b95555001","userId":"37628410","userIp":"113.101.248.138","userName":"陈嘉怡","data":{"subjectName":"数学","subjectId":"020","rankChallengeTime
":"201709","rankScope":"2"}}
*/
if (jsonStr.indexOf(Constant.SPLIT_MAKER) == -1) { // --
logger.info("--split error--" + jsonStr);
return;
}
//AutoActLogParseUtil autoActLogParseUtil = new AutoActLogParseUtil();
jsonStr = jsonStr.split(Constant.SPLIT_MAKER,2)[1]; // 得到要解析的字符串
jsonStr = jsonStr.replaceAll("\\\\\\\\n", " ");
jsonStr = jsonStr.replaceAll("\\\\n", " ");
jsonStr = jsonStr.replaceAll("\n", " ");
jsonStr = jsonStr.replaceAll("\\\\\\\\r", " ");
jsonStr = jsonStr.replaceAll("\\\\r", " ");
jsonStr = jsonStr.replaceAll("\r", " ");
/**
{
"actId":"1712012",
"classId":"f17ee6fd1fcb4f1fb3e1688f5cd364bd",
"className":"九年级(1)班",
"countyId":"445381",
"countyName":"罗定市",
"gradeId":"9",
"originCode":"4",
"role":"STUDENT",
"schoolId":"18738",
"schoolName":"罗定第二中学",
"time":1504281617255,
"userIcon":"d355db9797f854e1805e653b95555001",
"userId":"37628410",
"userIp":"113.101.248.138",
"userName":"陈嘉怡",
"data":{
"subjectName":"数学",
"subjectId":"020",
"rankScope":"2",
"rankChallengeTime":"201709"
}
}
*/
// logger.info("解析数据为：" + jsonStr);
JSONObject jsonObject = null;
try {
jsonObject = new JSONObject(jsonStr);
} catch (Exception e) {
logger.info("失败，无效的json格式；解析数据为：" + jsonStr);
return;
}
String actId ="";
if(jsonObject.has("actId")) {
actId = jsonObject.get("actId").toString();
if (actId.length() < 1) {
logger.info("失败，没有找到actID，解析数据为：" + jsonStr);
return;
}
}
else
{
logger.info("失败，打点数据没有actid key，解析数据为：" + jsonStr);
}
// logger.info("开始解析，解析数据为：" + jsonStr);
Object jsonObj = JSON.parse(jsonStr);
String sbff = AutoActLogParseUtil.jsonParse(jsonObj, actId, actMap); // 行数据 ， 里面代码写的太乱 实在看不懂 .... 预先建表的列顺序和json的行的列的存储顺序一致
if (StringUtils.isBlank(sbff)) {
logger.info("解析失败，解析数据为：" + jsonStr);
return;
}
Text outKey = new Text();
Text outValue = new Text();
outKey.set(actId);
outValue.set(sbff);
context.write(outKey, outValue);
} catch (IOException e) {
logger.error("IO错误", e);
} catch (JSONException e) {
logger.error("JSON格式不对", e);
// e.printStackTrace();
} catch (InterruptedException e) {
logger.error("JSON格式不对", e);
}
}
}
public static class AutoActLogParseReducer extends Reducer<Text, Text, NullWritable, Text> {
private MultipleOutputs<NullWritable, Text> mos; // 输出类型和Reduce一致
@Override
protected void setup(Reducer<Text, Text, NullWritable, Text>.Context context)
throws IOException, InterruptedException {
mos = new MultipleOutputs<NullWritable, Text>(context);
}
@Override
protected void cleanup(
Reducer<Text, Text, NullWritable, Text>.Context context)
throws IOException, InterruptedException {
mos.close();
}
@Override
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
Text valText = new Text();
Text keyValue = new Text();
Iterator<Text> it = values.iterator(); // 一行行的数据
String keyStr = key.toString();
while (it.hasNext()) {
String[] uk = it.next().toString().split(Constant.MARK_LINE); // \002 行数据之间用 \002间隔
for (String tmpUk : uk) {
valText.set(tmpUk);
System.out.println("keyStr="+keyStr+"valText"+valText); // keyStr=2900011valText0eb688ff16674be3b1642346aa347a7629000114f527f0cd64e4bf8bfaceeda4d33049a二年级(4)班445381罗定市a07973c2a487424f801c35b68517e88b76131115xl_9604e109d44c416492d23f1cdeb856e1_v111["a6c15215335649e29ddf48224a9bbe7e"]010语文2openApi210.0.26.434STUDENT19026素龙街中心小学15353860243005c8f4e4149fcce05f6c29cb5094b45ef39180478113.101.249.141杨大立
/**
* // String namedOutput, K key, V value, String baseOutputPath, 如果baseOutputPath不包含文件分隔符“/”，那么输出的文件格式为baseOutputPath-r-nnnnn（name-r-nnnnn)；
* 如果写成 mos.write(keyStr, NullWritable.get(), valText, keyStr + "hello"); 输出结果为 /log_data/2017-10-27/1100001hello-r-00001这种的，
* 而实际需要的是 /log_data/2017-10-27/1100001/-r-00001
*
* 如果是 mos.write(keyStr, NullWritable.get(), valText, keyStr + "/hello"); ， 则输出结果为 /log_data/2017-10-27/1100001/hello-r-00001
*
* 如果是 mos.write(keyStr, NullWritable.get(), valText, keyStr + "/"); 则输出结果为 /log_data/2017-10-27/1100001/-r-00001
*/
mos.write(keyStr, NullWritable.get(), valText, keyStr + "/"); // 对应于 run方法中的 addNamedOutput里的参数2， 这里必须保持一致 否则报Named ouput 'xxx' not defined的错
}
}
}
}
/**
* 来了新埋点，会产生新的建表语句，eg:
create external table stg_log_1900039
(uuid String,
actId String,
...
)COMMENT 'log'
PARTITIONED BY (day STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\001'
LINES TERMINATED BY '\n'
STORED AS TEXTFILE
LOCATION '/log_data/stg_log_1900039';
这里先去 /log_data/里找到所有符合 stg_log_1900039的表，解析得到所有埋点， 组品成 1900039@1900038这种方式返回
* @param hdfs
* @param path
* @param conf
* @param tableType
* @return
*/
public static String getOutPutName(FileSystem hdfs, Path path, Configuration conf,String tableType) { // fs, /log_data/ , conf , stg_log_
StringBuffer sb = new StringBuffer();
try {
if (hdfs == null || path == null) {
return sb.toString();
}
//获取文件列表
FileStatus[] listStatus = hdfs.listStatus(path);
Path[] listPath = FileUtil.stat2Paths(listStatus);
for (Path p : listPath) {
String fileNamePath = p.toString();//获得带路径的文件名字符串
String fileName = p.getName();//获取文件名 eg stg_log_1100001
String regex=tableType+"\\d{7}";
int fileNameLength=tableType.length()+7; // 15
int index=tableType.split("_").length; // 2
// if (flag==1) {
// regex = "stg_log_\\d{7}";//stg_log_1400004
// fileNameLength=15;
// index=2;
//
// }
// else {
// regex = "stg_log_class_perform_\\d{7}";//stg_log_1400004
// fileNameLength=29;
// index=4;
// }
// logger.info("regex="+regex+"and"+"filenameLength="+fileNameLength);
//logger.info("fileNamePath:"+fileNamePath+" fileName"+fileName);
Pattern pattern = Pattern.compile(regex);
if (fileName.length() == fileNameLength) {
// logger.info("filename7=" + fileName);
Matcher matcher = pattern.matcher(fileName);
if (matcher.matches()) {
//找到满足的文件，做下一步处理
if (!StringUtils.isBlank(sb.toString())) {
sb.append(Constant.MARK_AITE); // @
}
String str = matcher.group(); //
sb.append((str.split(Constant.MARK_XHX))[index]);//2,4
// logger.info("str.value is:" + str + " sb.value is " + sb);
}
//logger.info(" sb.value is " + sb); // 将需要解析的埋点 1100001@1400004 拼接成这样
}
}
} catch (IOException e) {
e.printStackTrace();
}
return sb.toString();
}
}
}
相关推荐
总之，通过使用Hadoop提供的`MultipleOutputs`工具类，可以在一个Reducer中实现生成多个输出文件的功能，这对于提高数据处理灵活性和效率具有重要意义。开发者需要根据具体的业务需求合理设计数据处理逻辑，同时也要...
在上面的案例中，我们使用Hadoop MultipleOutputs输出到多个文件中，每个文件对应一个类目。这样，可以将输出结果按照类目分类，方便后续处理。 Hadoop MultipleOutputs输出到多个文件中的实现方法可以满足我们对...
`MultipleOutputs`是Hadoop提供的一个工具类，它允许我们在Reducer中创建多个输出流，并分别指定它们的名称和路径。这使得我们可以在处理数据时，根据业务需求灵活地组织输出结果。 使用`MultipleOutputs`的基本...
在使用MapReduce框架时，经常需要处理输出数据，这时可能会遇到需要将输出分散到多个文件中的需求，这就是Hadoop MapReduce多输出功能的用途。Hadoop MapReduce多输出的功能主要由MultipleOutputFormat类及其相关类...
- **MultipleInputs/MultipleOutputs**：Hadoop API提供的工具类，用于一个Job处理多个输入源或产生多个输出结果。 3. **参数传递**： - **JobConf**：每个Job都有自己的JobConf对象，可以通过设置conf属性将参数...
2. **MultipleOutputs**：允许多个输出文件，便于处理不同类型的输出。 3. **Mapper/Reducer性能调优**：合理设置内存大小、槽位数量、并行度等参数。 通过以上介绍，你应该对Java MapReduce编程有了基本的认识。...
这可以通过Apache Hadoop的MultipleOutputs类实现，它允许数据流到多个输出路径，满足了这一需求。在Hadoop环境中，数据通常以分布式文件系统（如HDFS）的形式存储，并通过MapReduce进行处理。 项目开发中，Maven是...
### MapReduce中多文件输出的使用 在Hadoop MapReduce中处理大数据时，经常会遇到需要将输出结果分成多个文件的情况。这可能是为了便于后续的数据处理或存储，或者是因为业务需求。MapReduce中的多文件输出机制提供...
- 创建`Driver`类，设置输入输出路径，配置mapper和reducer，以及执行Job。 6. **Hadoop输入输出格式**： - Hadoop允许自定义输入输出格式，以适应不同类型的数据源和目标。 - 如`TextInputFormat`和`...
以及MapReduce编程模型的多个应用，包括求平均数、Join操作、TopK算法、二次排序，并涉及自定义InputFormat、OutputFormat和shuflle阶段的应用，如Partitioner、文件合并、MultipleOutPuts等。每个案例均独立封装在...
但使用`out`参数，我们可以让方法返回多个值。下面的示例展示了如何通过`out`参数返回两个值： ```csharp public static void MultipleOutputs(out string name, out int age) { name = "John Doe"; age = 30; } ...
项目中有一个特殊需求，即清洗后的数据需要按照设备类型（iOS、Android和其他）分别输出到不同的文件夹，这可以通过Hadoop的MultipleOutputs功能实现。 Maven作为项目管理工具，在这个项目中也扮演了重要角色。它...
读者将了解到如何自定义Mapper类以处理特定的数据转换，并理解Map输出的中间键值对是如何被分区和排序的。 第4章：Reduce阶段 本章详细讲解了Reduce任务的实现，包括Shuffle、Reduce函数以及Combine函数的使用。...
此外，还会探讨高级话题，如Combiner、MultipleOutputs和新版本的MapReduce API（YARN和Mesos上的运行机制）。 除了Hadoop核心，书中还涵盖了Hadoop生态中的其他重要组件，如HBase（一个分布式的、支持列族的NoSQL...