hadoop之MapReduce调用R的一次失败的总结~(续三)

路还在前进。

虽然后台错误如故，其实还是有些惊喜的。

我HBASE导入数据的时候弄错了一个参数，导致实际导入量为原来的1/1500。

我想说的是，我最初升级HBASE版本的时候是以为成功了。

因为我在表数据量很小的时候，执行程序（Scan 全表）后台是没有“断开的管道”之类错误的。

而恢复之前数据量的表的时候，执行程序（Scan 全表）后错误又回来了。

难道这个错误和表的数据量有关？

考虑到我实际的业务是根据rowkey检索到自己需要的数据。所以，进一步改写了MapReduce如下，

package mytest;

import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;

public class MapRLanguage3 {

public static int mapNum = 12;

public static final String RDIR_H="hdfs://bd110:9000/user/hadoop/";

public static final String RDIR_L="/home/hadoop/yibei/R/";

public static class RMapper extends Mapper<Object, Text, NullWritable, NullWritable>{
HTable table = null;

public void setup(Context context
) throws IOException, InterruptedException {
Configuration hbaseConf = HBaseConfiguration.create();
table = new HTable(hbaseConf, "kpinfo");
}

public void cleanup(Context context
) throws IOException, InterruptedException {
table.close();
}

//每个文件只有一行
public void map(Object key, Text value, Context context ) throws IOException, InterruptedException {
String args[] = value.toString().split("\\|");
//先准备存入的文件名称 例如：[Map].cell.kpi
InputSplit inputSplit = context.getInputSplit();
String fileName = ((FileSplit)inputSplit).getPath().getName();
//获取数据存入文件中
String cellarr[] = args[2].split(",");
String kpiarr[] = args[3].split(",");
//获取开始时间
String datedir = context.getConfiguration().get("datedir");
for(int i=0;i<cellarr.length;i++){
//创建文件
PrintWriter pw[]= new PrintWriter[kpiarr.length];
for(int j=0;j<kpiarr.length;j++){
File ldir = new File(RDIR_L+datedir+"/");
ldir.mkdirs();
File lfile = new File(RDIR_L+datedir+"/"+fileName+"."+cellarr[i]+"."+kpiarr[j]);
lfile.createNewFile();
pw[j] = new PrintWriter(lfile);
}
Scan scan = new Scan(Bytes.toBytes(cellarr[i]+"_"+args[4].replaceAll("-", "").replaceAll(":", "").replace(" ", "")),
Bytes.toBytes(cellarr[i]+"_"+args[5].replaceAll("-", "").replaceAll(":", "").replace(" ", "")));
ResultScanner scanner = table.getScanner(scan);
for(Result result: scanner){
String rowkey = Bytes.toString(result.getRow());
for(Map.Entry<byte[], byte[]> entry : result.getFamilyMap("h".getBytes()).entrySet()){
String column = new String(entry.getKey());
for(int k = 0; k < kpiarr.length;k++){
if(column.equalsIgnoreCase(kpiarr[k])){
pw[k].println(rowkey.substring(rowkey.indexOf("_")+1)+","+new String(entry.getValue()));
break;
}
}
}
}
scanner.close();
for(int j = 0;j<pw.length;j++){
pw[j].close();
}
}
//文件路径传入给R
// Rengine re=new Rengine(new String[] { "--vanilla" }, false,null);
// if (!re.waitForR()) {
// System.out.println("Cannot load R");
// return;
// }
// re.eval("setwd('"+RDIR_L+"')");
// re.eval("source('main2.R')");
// String rcmd="kpi.forecastByDBWithTime('"+args[4]+"','"+args[5]+"','"+args[6]+"','"+args[7]+"','"+args[1]+"','" + args[0] +"','" + args[2] +"','" + args[3] + "')";
// re.eval(rcmd);
// re.end();
}
}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
String dbtbl = "stat_plan_result_"+args[0].trim();
String dbname = args[1].trim();
String cells = args[2].trim();
String kpis = args[3].trim();
String hbtime = args[4].trim();
String hetime = args[5].trim();
String fbtime = args[6].trim();
String fetime = args[7].trim();
//2种思路，1种设置多少Size拆分一个Map使用FileInputFormat.setMaxInputSplitSize()，另1种利用每个文件一个Map的特性处理它。按文件处理也可以使用job.setInputFormatClass();
// String cells="1,2,3,4,5,6,7,8,9,10,11,12,13,14";
String cellarr[] = cells.split(",");
String datedir = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
String currentDir =RDIR_H+datedir+"/in/";
String currentoutDir =RDIR_H+datedir+"/out/";
if(cellarr.length<=mapNum){
for(int i =0;i<cellarr.length;i++){
//输出文件
String filepath = currentDir+i;
FileSystem fs = FileSystem.get(URI.create(filepath), new Configuration());
FSDataOutputStream out = fs.create(new Path(filepath));
out.write((dbtbl+"|"+dbname+"|"+cellarr[i]+"|"+kpis+"|"+hbtime+"|"+hetime+"|"+fbtime+"|"+fetime).getBytes("UTF-8"));
out.close();
}
}else{
int s = cellarr.length/mapNum;
int y = cellarr.length%mapNum;
for(int i =0,j=y;i<mapNum;i++){
//2种分配方案，1种是循环每个cell，加入文件。另1种是计算出，每次分配的数量。
int cellNum=s;
if(j>0){
cellNum+=1;
j--;
}
String celldata="";
for(int k =0;k < cellNum; k++){
//每个Map的小区，从cellarr中获取。
int cellindex=k+i*s+(cellNum>s?i:y);
celldata+=cellarr[cellindex]+",";
}
String filepath = currentDir+i;
FileSystem fs = FileSystem.get(URI.create(filepath), new Configuration());
FSDataOutputStream out = fs.create(new Path(filepath));
out.write((dbtbl+"|"+dbname+"|"+celldata.substring(0,celldata.length()-1)+"|"+kpis+"|"+hbtime+"|"+hetime+"|"+fbtime+"|"+fetime).getBytes("UTF-8"));
out.close();
}
}
Job job = Job.getInstance(new Configuration());
job.getConfiguration().setInt("mapreduce.task.timeout", 0);//关闭超时
job.getConfiguration().setBoolean("mapreduce.map.speculative", false);//关闭推测执行
job.getConfiguration().setStrings("datedir", datedir);//设置一个目录
job.setJarByClass(MapRLanguage3.class);
job.setMapperClass(RMapper.class);
//job.setOutputKeyClass(NullWritable.class);//和setMapOutputKeyClass的区别？也许是没有reduce的时候指的是一样的，有reduce的时候是只reduce的输出？
//job.setOutputValueClass(NullWritable.class);//同上
job.setNumReduceTasks(0);//没有Rudeuce输出
job.setOutputFormatClass(NullOutputFormat.class);//不需要输出 可以关闭setOutputKeyClass和setOutputValueClass？
FileInputFormat.addInputPath(job,new Path(currentDir));
FileOutputFormat.setOutputPath(job,new Path(currentoutDir));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}

}


实际的测试结果，后台还是会有“断开的管道”的错误（错误量少了很多，甚至有节点上没有错误），好在结果能正常返回，而且返回的快而正确。
如果忽略后台的错误日志，好像一切都那么完美！

升级Hbase1.1.1之后，我还做了一件事，测试最初的MapReduce（R通过thrift连接HBASE获取数据）

实测中，性能虽然感觉和JAVA端无法相比，要慢的多的多！但最关心的是以前的超时异常没有了。

不过新的麻烦又再次出现，就是程序结束后，再次调用这个MapReduce。

有时候会抛出如下异常

2015-07-22 15:49:01 cell:839 kpi:R99_UL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (TApplicationException) Internal error processing scannerOpenWithScan

2015-07-22 15:49:01 cell:842 kpi:R99_DL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (TApplicationException) Internal error processing scannerOpenWithScan

2015-07-22 15:49:01 cell:842 kpi:R99_UL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (TApplicationException) Internal error processing scannerOpenWithScan

2015-07-22 15:49:01 cell:856 kpi:R99_UL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (TApplicationException) Internal error processing scannerOpenWithScan

2015-07-22 15:49:01 cell:832 kpi:R99_UL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (TApplicationException) Internal error processing scannerOpenWithScan

2015-07-22 15:49:01 hbtime:2014-07-01 hetime:2014-10-01 fbtime:2015-07-10 fetime:2015-07-15 dbname:deeplan dbtbl:stat_plan_result_619 cells:853,852,838,839,842 kpis:R99_DL_USER,R99_UL_USER kpi.forecastByDBWithTime end!

2015-07-22 15:49:01 cell:849 kpi:R99_UL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (IOError) Default TException.

2015-07-22 15:49:01 cell:851 kpi:R99_DL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (IOError) Default TException.

2015-07-22 15:49:01 cell:851 kpi:R99_UL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (IOError) Default TException.

2015-07-22 15:49:01 cell:854 kpi:R99_DL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (IOError) Default TException.

2015-07-22 15:49:01 cell:854 kpi:R99_UL_USER is error Error in hb.scan.ex("kpinfo", filterstring = filterstring): rhbase<hbScannerOpenFilterEx>:: (IOError) Default TException.

他们出现的时间经常惊人的一致，估计是某个thrift方面瓶颈造成的吧。
去检索资料，貌似就下面这段话有用~

\code{rhbase<hbScannerOpenFilterEx>:: (TException) No more data to read.} This may have
the effect of breaking the connection between R and the Thrift server. To overcome this
issue, you will have to re-initalize the connection (i.e. call \code{hb.init} again).

考虑到rhbase本身就是一个没有被官方收录的包，实测中性能已经很成问题，我已经打算放弃。
那就不想多耗时间去分析，我能知道的就是，出现这个问题的时候，把thrift重启下就能解决了。