萬法歸宗之Hadoop編程無界限

時間 2019-12-06

標籤 hadoop 編程界限欄目 Hadoop 简体版

原文原文鏈接

記錄下，散仙今天的工做以及遇到的問題和解決方案，俗話說，好記性不如爛筆頭，寫出來文章，供你們參考，學習和點評，進步，纔是王道，廢話很少說，下面切入主題：

先介紹下需求：

散仙要處理多個相似表的txt數據，固然只有值，列名什麼的所有在xml裏配置了，而後加工這些每一個表的每一行數據，生成特定的格式基於ASCII碼1和ASCII碼2做爲分隔符的一行數據，ASCII2做爲字段名和字段值的分隔符，ASCII1做爲字段和字段之間的分隔符，每解析一個txt文件時，都要獲取文件名，而後與xml中的schema信息映射並找到對應位置的值，它的列名，前提是，這些的txt的內容位置，是固定的，而後咱們知道它每一行屬於哪一個表結構的映射，由於這些映射信息是提早配置在xml中的，以下圖：

固然相似這樣的結構有20個左右的表文件，到時候，咱們的數據方，會給咱們提供這些txt文件，而後散仙須要加工成特定的格式，而後寫入HDFS，由咱們的索引系統使用MapReduce批量建索引使用。

原本想直接用java寫個單機程序，串行處理，而後寫入HDFS，後來一想假如數據量比較大，串行程序還得改爲多線程並行執行，這樣改來改去，倒不如直接使用MapReduce來的方便

ok，說幹就幹，測試環境已經有一套CDH5.3的hadoop2.5集羣，直接就在eclipse進行開發和MapReduce程序的調試，反正也很久也沒手寫MapReduce了，前段時間，一直在用Apache Pig分析數據，此次處理的邏輯也不復雜，就再寫下練練手， CDH的集羣在遠程的服務器上，散仙本機的hadoop是Apache Hadoop2.2的版本，在使用eclipse進行開發時，也沒來得及換版本，理論上最好各個版本，不一樣發行版，之間對應起來開發比較好，這樣通常不會存在兼容性問題，但散仙此次就懶的換了，由於CDH5.x以後的版本，是基於社區版的Apache Hadoop2.2之上改造的，接口應該大部分都一致，固然這只是散仙猜測的。

（1）首先，散仙要搞定的事，就是解析xml了，在程序啓動以前須要把xml解析，加載到一個Map中，這樣在處理每種txt時，會根據文件名來去Map中找到對應的schma信息，解析xml，散仙直接使用的jsoup，具體爲啥，請點擊散仙這篇
http://qindongliang.iteye.com/blog/2162519文章，在這期間遇到了一個比較蛋疼的問題，簡直是一個bug，最先散仙定義的xml是每一個表，一個table標籤，而後它下面有各個property的映射定義，可是在用jsoup的cssQuery語法解析時，發現老是解析不出來東西，按照之前的作法，是沒任何問題的，此次簡直是開玩笑了，後來就是各類搜索，測試，最後才發現，將table標籤，換成其餘的任何標籤都無任何問題，具體緣由，散仙還沒來得及細看jsoup的源碼，猜想table標籤應該是一個關鍵詞什麼的標籤，在解析時會和html的table衝突，因此在xml解析中失效了，花了接近2個小時，求證，檢驗，終於搞定了這個小bug。

（2）搞定了這個問題，散仙就開始開發調試MapReduce版的處理程序，這下面臨的又一個問題，就是如何使用Jsoup解析存放在HDFS上的xml文件，有過Hadoop編程經驗的人，應該都知道，HDFS是一套分佈式的文件系統，與咱們本地的磁盤的存儲方式是不同的，好比你在正常的JAVA程序上解析在C:\file\t.tx或者在linux上/home/user/t.txt，所編寫的程序在Hadoop上是沒法使用的，你得使用Hadoop提供的編程接口獲取它的文件信息，而後轉成字符串以後，再給jsoup解析。

（3）ok，第二個問題搞定以後，你得編寫你的MR程序，處理對應的txt文本，並且保證不一樣的txt裏面的數據格式，所獲取的scheaml是正確的，因此在map方法裏，你要獲取固然處理文件的路徑，而後作相應判斷，在作對應處理。

（4）很好，第三個問題搞定以後，你的MR的程序，基本編寫的差很少了，下一步就改考慮如何提交到Hadoop的集羣上，來調試程序了，因爲散仙是在Win上的eclipse開發的，因此這一步可能遇到的問題會不少，並且加上，hadoop的版本不一致與發行商也不一致，出問題也純屬正常。

這裏多寫一點，通常建議你們不要在win上調試hadoop程序，這裏的坑很是多，若是能夠，仍是建議你們在linux上直接玩，下面說下，散仙今天又踩的坑，關於在windows上調試eclipse開發, 運行Yarn的MR程序，散仙之前也記錄了文章，感興趣者，能夠點擊這個連接
http://qindongliang.iteye.com/blog/2078452地址。

（5）提交前，是須要使用ant或maven或者java自帶的導出工具，將項目打成一個jar包提交的，這一點你們須要注意下，最後測試得出，Apache的hadoop2.2編寫的MR程序，是能夠直接向CDH Hadoop2.5提交做業的，可是因爲hadoop2.5中，使用google的guice做爲了一個內嵌的MVC輕量級的框架，因此在windows上打包提交時，須要引入額外的guice的幾個包，截圖以下：

上面幾步搞定後，打包整個項目，而後運行成功，過程以下：

Java代碼

輸出路徑存在，已刪除！
2015-04-08 19:35:18,001 INFO [main] client.RMProxy (RMProxy.java:createRMProxy(56)) - Connecting to ResourceManager at /172.26.150.18:8032
2015-04-08 19:35:18,170 WARN [main] mapreduce.JobSubmitter (JobSubmitter.java:copyAndConfigureFiles(149)) - Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
2015-04-08 19:35:21,156 INFO [main] input.FileInputFormat (FileInputFormat.java:listStatus(287)) - Total input paths to process : 2
2015-04-08 19:35:21,219 INFO [main] mapreduce.JobSubmitter (JobSubmitter.java:submitJobInternal(394)) - number of splits:2
2015-04-08 19:35:21,228 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - user.name is deprecated. Instead, use mapreduce.job.user.name
2015-04-08 19:35:21,228 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.jar is deprecated. Instead, use mapreduce.job.jar
2015-04-08 19:35:21,228 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - fs.default.name is deprecated. Instead, use fs.defaultFS
2015-04-08 19:35:21,229 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.reduce.tasks is deprecated. Instead, use mapreduce.job.reduces
2015-04-08 19:35:21,229 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.mapoutput.value.class is deprecated. Instead, use mapreduce.map.output.value.class
2015-04-08 19:35:21,230 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapreduce.map.class is deprecated. Instead, use mapreduce.job.map.class
2015-04-08 19:35:21,230 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.job.name is deprecated. Instead, use mapreduce.job.name
2015-04-08 19:35:21,230 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapreduce.inputformat.class is deprecated. Instead, use mapreduce.job.inputformat.class
2015-04-08 19:35:21,230 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.input.dir is deprecated. Instead, use mapreduce.input.fileinputformat.inputdir
2015-04-08 19:35:21,230 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.output.dir is deprecated. Instead, use mapreduce.output.fileoutputformat.outputdir
2015-04-08 19:35:21,230 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapreduce.outputformat.class is deprecated. Instead, use mapreduce.job.outputformat.class
2015-04-08 19:35:21,231 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.map.tasks is deprecated. Instead, use mapreduce.job.maps
2015-04-08 19:35:21,233 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.mapoutput.key.class is deprecated. Instead, use mapreduce.map.output.key.class
2015-04-08 19:35:21,233 INFO [main] Configuration.deprecation (Configuration.java:warnOnceIfDeprecated(840)) - mapred.working.dir is deprecated. Instead, use mapreduce.job.working.dir
2015-04-08 19:35:21,331 INFO [main] mapreduce.JobSubmitter (JobSubmitter.java:printTokens(477)) - Submitting tokens for job: job_1419419533357_5012
2015-04-08 19:35:21,481 INFO [main] impl.YarnClientImpl (YarnClientImpl.java:submitApplication(174)) - Submitted application application_1419419533357_5012 to ResourceManager at /172.21.50.108:8032
2015-04-08 19:35:21,506 INFO [main] mapreduce.Job (Job.java:submit(1272)) - The url to track the job: http://http://dnode1:8088/proxy/application_1419419533357_5012/
2015-04-08 19:35:21,506 INFO [main] mapreduce.Job (Job.java:monitorAndPrintJob(1317)) - Running job: job_1419419533357_5012
2015-04-08 19:35:33,777 INFO [main] mapreduce.Job (Job.java:monitorAndPrintJob(1338)) - Job job_1419419533357_5012 running in uber mode : false
2015-04-08 19:35:33,779 INFO [main] mapreduce.Job (Job.java:monitorAndPrintJob(1345)) - map 0% reduce 0%
2015-04-08 19:35:43,885 INFO [main] mapreduce.Job (Job.java:monitorAndPrintJob(1345)) - map 100% reduce 0%
2015-04-08 19:35:43,902 INFO [main] mapreduce.Job (Job.java:monitorAndPrintJob(1356)) - Job job_1419419533357_5012 completed successfully
2015-04-08 19:35:44,011 INFO [main] mapreduce.Job (Job.java:monitorAndPrintJob(1363)) - Counters: 27
File System Counters
FILE: Number of bytes read=0
FILE: Number of bytes written=166572
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
HDFS: Number of bytes read=47795
HDFS: Number of bytes written=594
HDFS: Number of read operations=12
HDFS: Number of large read operations=0
HDFS: Number of write operations=4
Job Counters
Launched map tasks=2
Data-local map tasks=2
Total time spent by all maps in occupied slots (ms)=9617
Total time spent by all reduces in occupied slots (ms)=0
Map-Reduce Framework
Map input records=11
Map output records=5
Input split bytes=252
Spilled Records=0
Failed Shuffles=0
Merged Map outputs=0
GC time elapsed (ms)=53
CPU time spent (ms)=2910
Physical memory (bytes) snapshot=327467008
Virtual memory (bytes) snapshot=1905754112
Total committed heap usage (bytes)=402653184
File Input Format Counters
Bytes Read=541
File Output Format Counters
Bytes Written=594
true

最後附上核心代碼，以做備忘：
（1）Map Only做業的代碼：

Java代碼

package com.dhgate.search.rate.convert;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FilenameFilter;
import java.io.IOException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.dhgate.parse.xml.tools.HDFSParseXmlTools;
import com.sun.xml.bind.v2.schemagen.xmlschema.Import;
/**
* 加工處理數據格式
*
* @author qindongliang 2015年04月07日
*
* **/
public class StoreConvert {
//log4j記錄
static Logger log=LoggerFactory.getLogger(StoreConvert.class);
/**
* 轉換支持的格式
*
* **/
private static class FormatMapper extends Mapper<LongWritable, Text, NullWritable, Text>{
@Override
protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
String filename = ((FileSplit) context.getInputSplit()).getPath().getName().split("\\.")[0];
//System.out.println("文件名是： "+filename);
//log.info("讀取的文件名是: "+filename);
String vs[]=value.toString().split(",");
if(HDFSParseXmlTools.map.get(filename)!=null){
Map<String, String> m=HDFSParseXmlTools.map.get(filename);
StringBuffer sb=new StringBuffer();
for(int i=0;i<vs.length;i++){
//字段\2值
if(i==vs.length-1){
sb.append(m.get(i+"")).append("\2").append(vs[i]);
}else{
sb.append(m.get(i+"")).append("\2").append(vs[i]).append("\1");
}
}
context.write(NullWritable.get(), new Text(filename+" == "+sb.toString()));
}
}
}
public static void main(String[] args) throws Exception {
// System.setProperty("HADOOP_USER_NAME", "root");
Configuration conf=new Configuration();
// getConf(conf);
conf.set("mapreduce.job.jar", "searchrate.jar");
conf.set("fs.defaultFS","hdfs://172.21.50.108:8020");
conf.set("mapreduce.framework.name", "yarn");
conf.set("mapred.remote.os", "Linux");
conf.set("yarn.resourcemanager.scheduler.address", "172.21.50.108:8030");
conf.set("yarn.resourcemanager.address", "172.21.50.108:8032");
// System.exit(0);
Job job=Job.getInstance(conf, "formatdata");
job.setJarByClass(StoreConvert.class);
// System.out.println("模式： "+conf.get("mapreduce.jobtracker.address"));;
job.setMapperClass(FormatMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setNumReduceTasks(0);//Map Only做業
String path = "/tmp/qin/out";
FileSystem fs = FileSystem.get(conf);
Path p = new Path(path);
if (fs.exists(p)) {
fs.delete(p, true);
System.out.println("輸出路徑存在，已刪除！");
}
FileInputFormat.setInputPaths(job, "/tmp/qin/testfile/");
FileOutputFormat.setOutputPath(job, p);
System.out.println(job.waitForCompletion(true));
}
}

使用解析HDFS上xml文件的代碼：

Java代碼

package com.dhgate.parse.xml.tools;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Created by qindongliang on 15-4-6.
* 大數據交流羣：415886155
*/
public class HDFSParseXmlTools {
private final static Logger log= LoggerFactory.getLogger(HDFSParseXmlTools.class);
//存儲元數據信息
public static Map<String, Map<String, String>> map=new HashMap<String, Map<String,String>>();
static Configuration conf=new Configuration();
static FileSystem fs=null;
static{
log.info("初始化加載mapping.xml開始.......");
try{
conf.set("fs.defaultFS","hdfs://172.21.50.108:8020/");
fs=FileSystem.get(conf);//獲取conf對象
Path xml =new Path("/tmp/qin/mapping.xml");//讀取HDFS的xml文件
BufferedReader br=new BufferedReader(new InputStreamReader(fs.open(xml)));//獲取輸入流
StringBuffer sb=new StringBuffer();//聲明一個buffer對象用來存儲xml文件內容
String line;
line=br.readLine();//讀取第一行
sb.append(line);//追加到StringBuffer中
while (line != null){
line=br.readLine();//循環讀取
sb.append(line);//循環追加
}
// System.out.println(sb.toString());
br.close();//釋放資源
Document d=Jsoup.parse(sb.toString(),"UTF-8");//解析xml
Set<String> set=new HashSet<String>();//排除，不須要解析的文件
List<Element> excludes=d.select("exclude");
for(Element ee:excludes){
set.add(ee.text().trim());
}
List<Element> tables=d.select("type");
for(Element t:tables){
String num=t.attr("num");
String name=t.attr("name");
String indexname=t.attr("indexname");
if(set.contains(name)){
log.info("跳過的表名："+name);
continue;
}
// System.out.println(" 序號: "+num+" 表名： "+name+" 索引名: "+indexname);
Map<String, String> data=new TreeMap<String, String>();
for(Element s:t.select("map")){
// System.out.println("----------------------"+s.attr("pos")+" "+s.attr("field")+" "+s.attr(""));
String pos=s.attr("pos");//位置信息
String field=s.attr("field");//索引字段名
data.put(pos, field);
}
map.put(name, data);//將此表名對應的映射信息存儲到map裏
}
}catch(Exception e){
//e.printStackTrace();
log.error("加載映射文件異常!",e);
}
}
public static void parseXml()throws Exception{
}
public static void main(String[] args) throws Exception{
System.out.println();
for(Entry<String, Map<String, String>> m:map.entrySet()){
System.out.println("表名："+m.getKey());
// for(Entry<String, String> me:m.getValue().entrySet()){
// System.out.println(me.getKey()+" "+me.getValue());
// }
// System.out.println("==================================================");
}
}
}

項目結構以下圖：

Ant的打包腳本以下：

Java代碼

<project name="${component.name}" basedir="." default="jar">
<property environment="env"/>
<property name="hadoop.home" value="E:/hadooplib"/>
<property name="jar.name" value="searchrate.jar"/>
<path id="project.classpath">
<fileset dir="lib">
<include name="*.jar" />
</fileset>
<fileset dir="${hadoop.home}">
<include name="**/*.jar" />
</fileset>
</path>
<target name="clean" >
<delete dir="bin" failonerror="false" />
<mkdir dir="bin"/>
</target>
<target name="build" depends="clean">
<echo message="${ant.project.name}: ${ant.file}"/>
<javac destdir="bin" encoding="utf-8" debug="true" includeantruntime="false" debuglevel="lines,vars,source">
<src path="src"/>
<exclude name="**/.svn" />
<classpath refid="project.classpath"/>
</javac>
<copy todir="bin">
<fileset dir="src">
<include name="*config*"/>
</fileset>
</copy>
</target>
<target name="jar" depends="build">
<copy todir="bin/lib">
<fileset dir="lib">
<include name="**/*.*"/>
</fileset>
</copy>
<copy todir="bin/lib">
<fileset dir="${hadoop.home}">
<include name="**/*.*"/>
</fileset>
</copy>
<path id="lib-classpath">
<fileset dir="lib" includes="**/*.jar" />
</path>
<pathconvert property="my.classpath" pathsep=" " >
<mapper>
<chainedmapper>
<flattenmapper />
<globmapper from="*" to="lib/*" />
</chainedmapper>
</mapper>
<path refid="lib-classpath" />
</pathconvert>
<jar basedir="bin" destfile="${jar.name}" >
<include name="**/*"/>
<manifest>
<attribute name="Class-Path" value="${my.classpath}" />
</manifest>
</jar>
</target>
</project>

至此，咱們以及完成了，這個小項目的開發，最終迴歸當生產環境上，咱們是須要打成jar包，在linux上定時執行的，直接使用linux環境來開發調試hadoop，遇到的問題會更少，雖然不推薦使用win直接開發hadoop程序，可是瞭解一些基本的方法和技巧，對咱們來講也是一件不錯的事情。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。