`
younglibin
  • 浏览: 1211745 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

eclipse 运行hadoop wordcount

阅读更多

给大家 一个建议 如果使用1.XX的版本hadoop  建议大家严格按照  以下的  第一条 中的博文的版本安装,可以是单机或者伪分布式,主要是因为,hadoop版本和 eclipse插件版本如果不一致会带来很多问题,解决起来比较麻烦,

先说明一下  此次搭建 hadoop  运行示例 ,过程中参考的文章:

1. eclipse 搭建 hadoop环境:

http://www.cnblogs.com/xia520pi/archive/2012/05/20/2510723.html

2.运行hadoop是遇到的问题:

  第一个问题:XX.XX.XXX.WordCount$TokenizerMapper   这个主要是因为 插件版本不一致的问题,但是目前我没有解决,我用的还是1.2.1的 有源码 ,查了一些资料 也手工打了一个  eclipse-plugin 但是最终还是不好使,打eclipse-plugins 的 可以参考  利用hadoop源码 打plugins的文章(我打了包但是还是不好用),绕过这个问题的方法可以参考 (http://www.cnblogs.com/spork/archive/2010/04/21/1717592.html  文章有几篇 写的不错,可以都看看也算都hadoop执行过程的一个了解) ,这个最终解决办法是模拟了 hadoop将本地文件打成jar 的过程

  以上问题解决 按照网上的文章应该是已经可以运行了 ,但是我这又遇到了 其他问题:

问题二:

eclipse 中运行 org.apache.hadoop.mapreduce.lib.input.InvalidInputException

    执行  wordount 的时候,配置input文件找不到:明明就在那里防止相对路径绝对路径都试过了 还是不行,最后想着 hadoop在命令行执行 的话,是 将本地文件上传到  hadoop中的,但是我运行时候每次都报的的是本地文件找不到,所以应该写远程文件地址,参开 执行成功的hadoop文件发现文件地址为:

mapred.output.dirhdfs://172.16.236.11:9000/user/libin/output/wordcount-ec

 

所以就修改了本地代码程序:

		FileInputFormat.addInputPath(job, new Path("hdfs://172.16.236.11:9000"
				+ File.separator + otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(
				"hdfs://172.16.236.11:9000" + File.separator + otherArgs[1]));

 

这样 本地执行的代码就可以 在hadoop服务端执行了。

 

这里还有一个问题 就是,如果每次测试都需要将本地文件上传到 hadoop服务端 ,好像有点麻烦,所以 可以考虑,每次addInputPath 的时候,在这个之前,先执行以下 类型 hadoop fs  -put 的 代码,将本地文件上传到  hadoop服务端,这样就不用每次手工上传文件到 服务器 参考 

 

http://younglibin.iteye.com/admin/blogs/1925109

 

贴一个完整的 示例程序吧:

 

package com.younglibin.hadoop.test;

import java.io.File;
import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
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 com.younglibin.hadoop.EJob;

public class WordCount {
	public static class TokenizerMapper extends
			Mapper<Object, Text, Text, IntWritable> {

		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			StringTokenizer itr = new StringTokenizer(value.toString());
			while (itr.hasMoreTokens()) {
				word.set(itr.nextToken());
				context.write(word, one);
			}
		}
	}

	public static class IntSumReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
		}
	}

	public static void main(String[] args) throws Exception {

		File jarFile = EJob.createTempJar("bin");
		EJob.addClasspath("/home/libin/software/hadoop/hadoop-1.2.1/conf");
		ClassLoader classLoader = EJob.getClassLoader();
		Thread.currentThread().setContextClassLoader(classLoader);

		Configuration conf = new Configuration();
		conf.set("mapred.job.tracker", "172.16.236.11:9001");
		args = new String[] { "/user/libin/input/libin", "/user/libin/output/wordcount-ec" };
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: wordcount <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf, "word count");
		job.setJarByClass(WordCount.class);
		((JobConf) job.getConfiguration()).setJar(jarFile.toString());
		job.setMapperClass(TokenizerMapper.class);
		job.setCombinerClass(IntSumReducer.class);
		job.setReducerClass(IntSumReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path("hdfs://172.16.236.11:9000"
				+ File.separator + otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(
				"hdfs://172.16.236.11:9000" + File.separator + otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

 

  

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.younglibin.hadoop;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Array;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
import java.util.jar.JarOutputStream;
import java.util.jar.Manifest;

public class EJob {

	private static ArrayList<URL> classPath = new ArrayList<URL>();

	/** Unpack a jar file into a directory. */
	public static void unJar(File jarFile, File toDir) throws IOException {
		JarFile jar = new JarFile(jarFile);
		try {
			Enumeration entries = jar.entries();
			while (entries.hasMoreElements()) {
				JarEntry entry = (JarEntry) entries.nextElement();
				if (!entry.isDirectory()) {
					InputStream in = jar.getInputStream(entry);
					try {
						File file = new File(toDir, entry.getName());
						if (!file.getParentFile().mkdirs()) {
							if (!file.getParentFile().isDirectory()) {
								throw new IOException("Mkdirs failed to create "
										+ file.getParentFile().toString());
							}
						}
						OutputStream out = new FileOutputStream(file);
						try {
							byte[] buffer = new byte[8192];
							int i;
							while ((i = in.read(buffer)) != -1) {
								out.write(buffer, 0, i);
							}
						} finally {
							out.close();
						}
					} finally {
						in.close();
					}
				}
			}
		} finally {
			jar.close();
		}
	}

	/**
	 * Run a Hadoop job jar. If the main class is not in the jar's manifest, then
	 * it must be provided on the command line.
	 */
	public static void runJar(String[] args) throws Throwable {
		String usage = "jarFile [mainClass] args...";

		if (args.length < 1) {
			System.err.println(usage);
			System.exit(-1);
		}

		int firstArg = 0;
		String fileName = args[firstArg++];
		File file = new File(fileName);
		String mainClassName = null;

		JarFile jarFile;
		try {
			jarFile = new JarFile(fileName);
		} catch (IOException io) {
			throw new IOException("Error opening job jar: " + fileName).initCause(io);
		}

		Manifest manifest = jarFile.getManifest();
		if (manifest != null) {
			mainClassName = manifest.getMainAttributes().getValue("Main-Class");
		}
		jarFile.close();

		if (mainClassName == null) {
			if (args.length < 2) {
				System.err.println(usage);
				System.exit(-1);
			}
			mainClassName = args[firstArg++];
		}
		mainClassName = mainClassName.replaceAll("/", ".");

		File tmpDir = new File(System.getProperty("java.io.tmpdir"));
		tmpDir.mkdirs();
		if (!tmpDir.isDirectory()) {
			System.err.println("Mkdirs failed to create " + tmpDir);
			System.exit(-1);
		}
		final File workDir = File.createTempFile("hadoop-unjar", "", tmpDir);
		workDir.delete();
		workDir.mkdirs();
		if (!workDir.isDirectory()) {
			System.err.println("Mkdirs failed to create " + workDir);
			System.exit(-1);
		}

		Runtime.getRuntime().addShutdownHook(new Thread() {
			public void run() {
				try {
					fullyDelete(workDir);
				} catch (IOException e) {
				}
			}
		});

		unJar(file, workDir);

		classPath.add(new File(workDir + "/").toURL());
		classPath.add(file.toURL());
		classPath.add(new File(workDir, "classes/").toURL());
		File[] libs = new File(workDir, "lib").listFiles();
		if (libs != null) {
			for (int i = 0; i < libs.length; i++) {
				classPath.add(libs[i].toURL());
			}
		}

		ClassLoader loader = new URLClassLoader(classPath.toArray(new URL[0]));

		Thread.currentThread().setContextClassLoader(loader);
		Class<?> mainClass = Class.forName(mainClassName, true, loader);
		Method main = mainClass.getMethod("main", new Class[] { Array.newInstance(
				String.class, 0).getClass() });
		String[] newArgs = Arrays.asList(args).subList(firstArg, args.length)
				.toArray(new String[0]);
		try {
			main.invoke(null, new Object[] { newArgs });
		} catch (InvocationTargetException e) {
			throw e.getTargetException();
		}
	}

	/**
	 * Delete a directory and all its contents. If we return false, the directory
	 * may be partially-deleted.
	 */
	public static boolean fullyDelete(File dir) throws IOException {
		File contents[] = dir.listFiles();
		if (contents != null) {
			for (int i = 0; i < contents.length; i++) {
				if (contents[i].isFile()) {
					if (!contents[i].delete()) {
						return false;
					}
				} else {
					// try deleting the directory
					// this might be a symlink
					boolean b = false;
					b = contents[i].delete();
					if (b) {
						// this was indeed a symlink or an empty directory
						continue;
					}
					// if not an empty directory or symlink let
					// fullydelete handle it.
					if (!fullyDelete(contents[i])) {
						return false;
					}
				}
			}
		}
		return dir.delete();
	}

	/**
	 * Add a directory or file to classpath.
	 * 
	 * @param component
	 */
	public static void addClasspath(String component) {
		if ((component != null) && (component.length() > 0)) {
			try {
				File f = new File(component);
				if (f.exists()) {
					URL key = f.getCanonicalFile().toURL();
					if (!classPath.contains(key)) {
						classPath.add(key);
					}
				}
			} catch (IOException e) {
			}
		}
	}

	/**
	 * Add default classpath listed in bin/hadoop bash.
	 * 
	 * @param hadoopHome
	 */
	public static void addDefaultClasspath(String hadoopHome) {
		// Classpath initially contains conf dir.
		addClasspath(hadoopHome + "/conf");

		// For developers, add Hadoop classes to classpath.
		addClasspath(hadoopHome + "/build/classes");
		if (new File(hadoopHome + "/build/webapps").exists()) {
			addClasspath(hadoopHome + "/build");
		}
		addClasspath(hadoopHome + "/build/test/classes");
		addClasspath(hadoopHome + "/build/tools");

		// For releases, add core hadoop jar & webapps to classpath.
		if (new File(hadoopHome + "/webapps").exists()) {
			addClasspath(hadoopHome);
		}
		addJarsInDir(hadoopHome);
		addJarsInDir(hadoopHome + "/build");

		// Add libs to classpath.
		addJarsInDir(hadoopHome + "/lib");
		addJarsInDir(hadoopHome + "/lib/jsp-2.1");
		addJarsInDir(hadoopHome + "/build/ivy/lib/Hadoop/common");
	}

	/**
	 * Add all jars in directory to classpath, sub-directory is excluded.
	 * 
	 * @param dirPath
	 */
	public static void addJarsInDir(String dirPath) {
		File dir = new File(dirPath);
		if (!dir.exists()) {
			return;
		}
		File[] files = dir.listFiles();
		if (files == null) {
			return;
		}
		for (int i = 0; i < files.length; i++) {
			if (files[i].isDirectory()) {
				continue;
			} else {
				addClasspath(files[i].getAbsolutePath());
			}
		}
	}

	/**
	 * Create a temp jar file in "java.io.tmpdir".
	 * 
	 * @param root
	 * @return
	 * @throws IOException
	 */
	public static File createTempJar(String root) throws IOException {
		if (!new File(root).exists()) {
			return null;
		}
		Manifest manifest = new Manifest();
		manifest.getMainAttributes().putValue("Manifest-Version", "1.0");
		final File jarFile = File.createTempFile("EJob-", ".jar", new File(System
				.getProperty("java.io.tmpdir")));

		Runtime.getRuntime().addShutdownHook(new Thread() {
			public void run() {
				jarFile.delete();
			}
		});

		JarOutputStream out = new JarOutputStream(new FileOutputStream(jarFile),
				manifest);
		createTempJarInner(out, new File(root), "");
		out.flush();
		out.close();
		return jarFile;
	}

	private static void createTempJarInner(JarOutputStream out, File f,
			String base) throws IOException {
		if (f.isDirectory()) {
			File[] fl = f.listFiles();
			if (base.length() > 0) {
				base = base + "/";
			}
			for (int i = 0; i < fl.length; i++) {
				createTempJarInner(out, fl[i], base + fl[i].getName());
			}
		} else {
			out.putNextEntry(new JarEntry(base));
			FileInputStream in = new FileInputStream(f);
			byte[] buffer = new byte[1024];
			int n = in.read(buffer);
			while (n != -1) {
				out.write(buffer, 0, n);
				n = in.read(buffer);
			}
			in.close();
		}
	}

	/**
	 * Return a classloader based on user-specified classpath and parent
	 * classloader.
	 * 
	 * @return
	 */
	public static ClassLoader getClassLoader() {
		ClassLoader parent = Thread.currentThread().getContextClassLoader();
		if (parent == null) {
			parent = EJob.class.getClassLoader();
		}
		if (parent == null) {
			parent = ClassLoader.getSystemClassLoader();
		}
		return new URLClassLoader(classPath.toArray(new URL[0]), parent);
	}

}

 

 

看这个比较迷茫的是参数的传递:

简单说明一下,这里map参数传递 在 FileInputFormat 指定的 ,所以在map方法中,中的value

就是经过FileInputFormat  处理的   ,在处理 createRecordReader 的时候,根据参数来判断使用哪一个子类处理,这里使用了TextInputFormat 

 

分享到:
评论
1 楼 gwgyk 2014-09-29  
关于楼主最后的问题,我想可能是这样:
InputFormat默认实现是TextInputFormat,而在TextInputFormat中,createRecordReader()方法中最后一行语句是:
return new LineRecordReader 

不知这样是否解决了你的问题

相关推荐

    hadoop wordcount 打包部署

    ### Hadoop WordCount项目打包与部署详解 #### 一、Hadoop WordCount...通过以上步骤,可以成功地在Hadoop环境下部署并运行WordCount程序。这不仅是一个基础的数据处理练习,也是深入了解Hadoop生态系统的重要一步。

    eclipse配置hadoop

    六、在 Eclipse 中运行 WordCount 程序 6.1 导入 WordCount WordCount 6.2 配置运行参数 Run As -&gt; Open Run Dialog... 选择 WordCount 程序,在 Arguments 中配置运行参数:/mapreduce/wordcount/input /...

    eclipse下运行wordcount

    在本文中,我们将深入探讨如何在Windows环境下使用Eclipse IDE远程连接到伪分布式Hadoop集群,并运行经典的WordCount示例程序。WordCount是Hadoop入门的标志性任务,它统计文本中单词出现的次数,这对于理解...

    基于Windows eclipse maven Hadoop 的WordCount源码

    总结来说,基于Windows、Eclipse、Maven和Hadoop实现WordCount源码的过程主要包括以下步骤:配置开发环境、创建Maven项目、编写MapReduce代码、打包成jar文件,以及在Hadoop集群上运行作业。通过这个过程,我们可以...

    eclipse hadoop插件安装 运行WordCount

    在本文中,我们将深入探讨如何在Eclipse IDE中安装Hadoop插件,并通过该插件运行WordCount示例,特别关注Windows环境下的配置。首先,我们需要了解Eclipse Hadoop插件的作用,它允许开发者在Eclipse环境中直接创建、...

    基于Eclipse的Hadoop应用开发环境配置

    基于 Eclipse 的 Hadoop 应用开发环境配置是指在 Eclipse 中配置 Hadoop 开发环境,以便开发和运行 Hadoop 应用程序。本节将对基于 Eclipse 的 Hadoop 应用开发环境配置进行详细介绍。 一、Hadoop 概述 Hadoop 是...

    eclipse连接hadoop所需要的hadoop.ddl和eclipse插件和hadoop运行案例

    以下是对"eclipse连接hadoop所需要的hadoop.ddl和eclipse插件和hadoop运行案例"这一主题的详细解释: 首先,让我们了解`hadoop.ddl`。DDL(Data Definition Language)通常指的是数据库中用于定义数据结构的语句,...

    eclipse运行hadoop2.7.3程序依赖的jar包

    为了在Eclipse中运行Hadoop 2.7.3程序,你需要确保正确配置了项目依赖并包含了所有必需的JAR包。以下是你需要知道的关键知识点: 1. **Hadoop环境搭建**:首先,你需要在Linux服务器上安装Hadoop 2.7.3。这通常包括...

    cygwin+eclipse搭建hadoop开发环境,运行wordcount

    总的来说,这些文档和资料将帮助你搭建一个完整的Hadoop开发环境,从安装Cygwin和Eclipse,到配置Hadoop环境,最后通过运行WordCount实例来验证你的环境是否正确配置。这不仅对于初学者来说是一次很好的学习体验,也...

    配置Eclipse连接Hadoop

    你可以通过"Run As" -&gt; "Hadoop Job"运行一个简单的MapReduce程序进行测试,比如WordCount。 - "单机测试 Hadoop.txt"可能是记录此类测试过程的文档,可以参照其中的步骤来检查配置是否正确。 5. **编写和调试...

    hadoop wordCount程序

    在 Eclipse 中,可以创建一个 Java 类,例如 `WordCount`,然后将 Hadoop 的 WordCount 源程序代码拷贝到该类中。该代码实现了 WordCount 程序的逻辑,包括 Map 任务和 Reduce 任务。 运行 WordCount 程序 要运行 ...

    Linux和Win下配置eclipse开发hadoop

    10. **运行和调试**: 在Eclipse中可以直接运行和调试Hadoop程序,通过Eclipse的Run As菜单选择Map/Reduce Job。 **二、Windows环境下配置Eclipse开发Hadoop** 1. **安装Java**: Windows上同样需要Java环境,下载并...

    基于eclipse的hadoop应用开发

    通过上述步骤,我们不仅完成了在Eclipse环境下配置Hadoop开发环境的任务,还成功实现了WordCount程序的开发与运行,这对于初学者来说是非常好的实践案例。未来可以在此基础上尝试更复杂的大数据处理任务,进一步探索...

    windows eclipse 运行wordcount连接linux hadoop2.8 NativeIO YARNRunner 完项目 源码

    在本项目中,我们主要关注的是如何在Windows环境下使用Eclipse IDE运行一个WordCount程序,该程序连接到Linux上的Hadoop 2.8集群,并利用NativeIO和YARNRunner进行分布式处理。以下是对这些关键概念的详细解释: 1....

    Ubuntu下开发Eclipse下的Hadoop应用

    本文档详细记录了如何在Ubuntu下安装Eclipse,并如何在Eclipse中安装Hadoop插件,最后运行程序。下面将对标题、描述、标签和部分内容进行详细解释和分析。 标题: Ubuntu下开发Eclipse下的Hadoop应用 本标题表明本...

    linux环境下运行Eclipse用来测试WordCount示例

    完成以上步骤后,就可以在Eclipse中创建新的Hadoop项目,并运行经典的WordCount示例了。通过这种方式,不仅可以验证Hadoop插件的正确安装,还能熟悉Eclipse IDE在Linux环境下对Hadoop的支持特性。 总结来说,本文...

    Eclipse中编译运行Hadoop-0.20.1源码

    在Eclipse中编译和运行Hadoop-0.20.1源码是理解Hadoop工作原理和进行源代码级调试的重要步骤。以下是一个详细的过程,涵盖了从环境准备到源码编译和运行的全部环节。 首先,确保你的开发环境满足必要的前提条件。在...

    windows平台使用hadoop hdfs文件进行中文分词的示例代码

    为了运行这个程序,你需要将《唐诗三百首》的文本文件上传到HDFS,并在Hadoop的配置中指定输入文件路径。在Eclipse中,通过右键点击项目,选择“Run As” &gt; “Hadoop Job”,设置好Job配置后提交作业。程序执行完成...

    基于Eclipse的Hadoop应用开发环境配置.docx

    运行项目时,首先在Eclipse中为`WordCount.java`创建一个运行配置,指定输入文件和预期输出文件的路径。如果遇到内存问题,可能需要调整Java虚拟机(JVM)的堆大小和永久代大小,通过在VM arguments中添加适当的设置...

Global site tag (gtag.js) - Google Analytics