从今天开始做一个自己的hadoop的例子，例子的目标是完成如下任务。
任务目标：
有一个informix数据库，其中有一个表有100万的数据，任务完成通过MapRed的方式将这100万数据导入到HDFS中，可以使用informix的分页sql达到并发查询数据库的目的。
任务开始：
1、编写MapRed的实现类起名称为InformixLoader。
大致代码如下
LOG.info("SqlMapper");
String url = context.getConfiguration().get("informix.url");
String[] str = value.toString().split("_");
long start = Long.parseLong(str[0]);
long length = Long.parseLong(str[1]);
Connection conn = null;
Statement st = null;
ResultSet rs = null;
String sql = "";
try {
Class.forName("com.informix.jdbc.IfxDriver");
conn = DriverManager.getConnection(url, "niosuser", "db");
st = conn.createStatement();
sql = "select skip " + start + " first " + length
+ " int_id from tpa_sts_cell_ne";
LOG.info("SqlMapper sql:" + sql);
rs = st.executeQuery(sql);
dataKey.set(value.toString());
Path file = new Path(StringUtils.unEscapeString(TMP_MAP_DIR
+ value.toString()));
FileSystem fs = file.getFileSystem(context.getConfiguration());
fs.createNewFile(file);
FSDataOutputStream output = fs.create(file);
LOG.info("SqlMapper createNewFile OK!");
while (rs.next()) {
String c1 = rs.getString(1)+"\n";
output.write(c1.getBytes());
output.flush();
}
output.close();
// fs.close();
data.set(value.toString());
context.write(dataKey, value);
LOG.info("SqlMapper OK!");
} catch (Exception e) {
throw new IOException(sql, e.fillInStackTrace());
} finally {
if (rs != null) {
try {
rs.close();
} catch (SQLException e) {
throw new IOException(e.fillInStackTrace());
}
}
if (st != null) {
try {
st.close();
} catch (SQLException e) {
throw new IOException(e.fillInStackTrace());
}
}
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
throw new IOException(e.fillInStackTrace());
}
}
}
2、编写InputFormater类
代码如下：
LOG.info("InputFormarter");
String url = context.getConfiguration().get("informix.url");
Connection conn = null;
Statement st = null;
ResultSet rs = null;
try {
Class.forName("com.informix.jdbc.IfxDriver");
conn = DriverManager.getConnection(url, "niosuser", "db");
st = conn.createStatement();
String sql = "select count(*) from tpa_sts_cell_ne";
rs = st.executeQuery(sql);
rs.next();
int count = rs.getInt(1);
List<InputSplit> splits = new ArrayList<InputSplit>();
int size = 50000;
int inv = count / size;
int last = count % size;
for (int i = 0; i < inv; i++) {
SqlSplit s = new SqlSplit(i * size, size);
splits.add(s);
}
if (last!=0){
SqlSplit s = new SqlSplit(inv * size, last);
splits.add(s);
}
return splits;
} catch (Exception e) {
throw new IOException(e.fillInStackTrace());
} finally {
if (rs != null) {
try {
rs.close();
} catch (SQLException e) {
throw new IOException(e.fillInStackTrace());
}
}
if (st != null) {
try {
st.close();
} catch (SQLException e) {
throw new IOException(e.fillInStackTrace());
}
}
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
throw new IOException(e.fillInStackTrace());
}
}
}
3、编写reducer类
大致代码如下
String keyStr = key.toString();
Path outFile = new Path(StringUtils.unEscapeString(TMP_RED_DIR
+ keyStr));
LOG.info("SqlReducer outfile:"+outFile.getName());
FileSystem outfs = outFile
.getFileSystem(context.getConfiguration());
outfs.createNewFile(outFile);
FSDataOutputStream output = outfs.create(outFile);
for (Text val : values) {
LOG.info("SqlReducer");
String str = val.toString();
LOG.info("file:"+str);
Path inputFile = new Path(StringUtils
.unEscapeString(TMP_MAP_DIR + str));
FileSystem fs = inputFile.getFileSystem(context
.getConfiguration());
FSDataInputStream input = fs.open(inputFile);
BufferedInputStream bi = new BufferedInputStream(input);
byte[] buffer=new byte[1024];
int length=bi.read(buffer);
while (length!=-1) {
if (length==1024){
output.write(buffer);
}else{
byte[] tmp=new byte[length];
for(int i=0;i<tmp.length;i++){
tmp[i]=buffer[i];
}
output.write(buffer);
}
length=bi.read(buffer);
}
bi.close();
input.close();
// fs.close();
output.flush();
}
output.close();
result.set(key.toString());
context.write(key, result);
4、编写outformat类
大致代码如下：
Path outFilePath = getDefaultWorkFile(context, "");
final FileSystem fs = outFilePath.getFileSystem(context
.getConfiguration());
final FSDataOutputStream output = fs.create(outFilePath);
return new RecordWriter<Text, Text>() {
@Override
public void close(TaskAttemptContext context)
throws IOException, InterruptedException {
output.flush();
output.close();
// fs.close();
}
@Override
public void write(Text key, Text value) throws IOException,
InterruptedException {
LOG.info("RecordWriter filename:"+value.toString());
Path file = new Path(StringUtils.unEscapeString(TMP_RED_DIR
+ value.toString()));
FileSystem fs = file.getFileSystem(context
.getConfiguration());
FSDataInputStream input = fs.open(file);
BufferedInputStream bi = new BufferedInputStream(input);
byte[] buffer=new byte[1024];
int length=bi.read(buffer);
while (length!=-1) {
if (length==1024){
output.write(buffer);
}else{
byte[] tmp=new byte[length];
for(int i=0;i<tmp.length;i++){
tmp[i]=buffer[i];
}
output.write(buffer);
}
length=bi.read(buffer);
}
bi.close();
input.close();
// fs.close();
}
};
5、编写启动代码
大致代码如下：
File jarFile = EJob.createTempJar("bin");
EJob.addClasspath("I:\\work\\hadoop\\hadoop\\hadoop-site.xml");
ClassLoader classLoader = EJob.getClassLoader();
Thread.currentThread().setContextClassLoader(classLoader);
args = new String[] { "/tmp/sqlloader10/" };
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args)
.getRemainingArgs();
if (otherArgs.length != 1) {
System.err.println("Usage: informixLoader <out>");
System.exit(2);
}
conf
.set("informix.url",
"jdbc:informix-sqli://10.0.2.36:8001/niosdb:INFORMIXSERVER=niosserver");
Job job = new Job(conf, "informix loader");
// And add this statement. XXX
((JobConf) job.getConfiguration()).setJar(jarFile.toString());
job.setInputFormatClass(InputFormarter.class);
job.setJarByClass(InformixLoader.class);
job.setMapperClass(SqlMapper.class);
job.setReducerClass(SqlReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setOutputFormatClass(OutputFormater.class);
// FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
5、详细分析后续说明
分享到：
相关推荐
hadoop入门例子wordcount
【描述】：“eclipse hadoop例子源代码 eclipse hadoop例子源代码” 描述中重复了标题的信息，强调了这是Eclipse环境下针对Hadoop的源代码实例。这通常意味着我们可以期待找到如何在Eclipse中配置Hadoop开发环境，...
Hadoop是Apache软件基金会开发的一个开源分布式计算框架，它允许在大规模集群中高效处理和存储海量数据。这个压缩包文件包含的"hadop实用案例"很可能是为了帮助初学者理解和应用Hadoop技术。以下是关于Hadoop的一些...
hadoop常用算法例子 Hadoop是云计算框架之一，提供了强大的数据处理能力。MapReduce是Hadoop的核心组件，用于处理大规模数据。下面是Hadoop常用的算法例子： 1. 基本MapReduce模式计数与求和 问题陈述：有许多...
Hadoop MapReduce是一种分布式计算框架，它允许在大型数据集上进行并行处理。这个例子项目是关于在单机环境中运行WordCount程序的，这是一个经典的MapReduce示例，用于统计文本文件中每个单词出现的次数。 首先，让...
在IT行业中，Hadoop是一个广泛使用的开源框架，它主要用于处理和存储海量数据。这个"**Hadoop简单应用案例**"涵盖了Hadoop生态系统中的多个关键组件，包括MapReduce、HDFS、Zookeeper以及Hive，这些都是大数据处理的...
【HADOOP案例及测试资料】是一份涵盖了Hadoop平台搭建、实例运行、源代码分析、测试问题以及基准测试的综合资料集。这份压缩包包含了多个文档，它们分别提供了不同方面的深入理解和实践指导。 首先，"Hadoop平台...
小例子是自己写的，目的是让自己熟悉一下如何在集群上运行一个mapreduce项目，大家可以参考此例子学习hadoop，对入门很有帮助。小例子是自己写的，目的是让自己熟悉一下如何在集群上运行一个mapreduce项目，大家可以...
在Java编程环境中，Hadoop是一个分布式计算框架，用于处理和存储海量数据。在这个"Java写的hadoop压缩worldcount例子"中，我们主要关注的是如何...通过分析和实践这个例子，开发者能够提升自己在大数据处理领域的技能。
Hadoop 是一种基于云计算的分布式计算框架，由 Apache 基金会在2002年发起，起源于 Apache Nutch 项目。它的核心是分布式文件系统 HDFS（Hadoop Distributed File System）和 MapReduce 计算模型。Hadoop 设计的目标...
3. **插件集成**：Eclipse通过插件机制扩展其功能，Hadoop2 插件就是这样的一个例子。它将Hadoop相关的开发工具集成了到Eclipse中，使开发者无需离开IDE即可进行Hadoop开发。 4. **主要功能**： - **项目创建**：...
这个例子展示了Hadoop的核心工作流程，包括Mapper和Reducer两个主要阶段。 首先，我们来了解一下MapReduce模型。MapReduce是一种编程模型，用于大规模数据集（大于1TB）的并行计算。它将大型任务分解为小部分，然后...
- 第一个Hadoop程序：通过简单的WordCount例子了解MapReduce的工作原理。 2. **Hadoop 集群搭建**： - 集群规划：如何根据硬件资源和预期负载来规划节点数量和类型。 - 配置文件详解：如core-site.xml, hdfs-...
在这个例子中，文件已经包含在 "centos6.5-hadoop-2.6.4.tar.gz" 中。 3. **解压Hadoop**：使用`tar`命令解压下载的文件，例如 `tar -zxvf centos6.5-hadoop-2.6.4.tar.gz`，这将在当前目录下创建一个名为 `hadoop-...
本资源包“hadoop学习例子”旨在帮助初学者理解并掌握Hadoop的核心概念和实际操作。以下是对其中涉及知识点的详细阐述： 1. **分布式文件系统HDFS（Hadoop Distributed File System）**：Hadoop的核心之一是HDFS，...
一个自己写的Hadoop MapReduce实例源码，网上看到不少网友在学习MapReduce编程，但是除了wordcount范例外实例比较少，故上传自己的一个。包含完整实例源码，编译配置文件，测试数据，可执行jar文件，执行脚本及操作...
亚历克斯·霍姆斯 (Alex Holmes)、 梁李印 需要下载了1和2之后，才能解压开。解压的时候，两本放在同一目录。 ...HADOOP硬实战 Hadoop是一个开源的...本书提供了结构良好且易于理解的例子，可用于应对你所遇到的问题。
《HadoopDemo：初识与实践》 HadoopDemo是一个专为Hadoop初学者设计的项目，它通过一系列实例展示了Hadoop的核心功能和使用方法。Hadoop，作为大数据处理领域的重要框架，以其分布式、容错性和扩展性著称，是理解和...
`mapreduce`目录下的文件可能包括了MapReduce的示例程序或者库文件，帮助用户理解并编写自己的MapReduce作业。 使用这些jar包，开发者可以编写应用程序，通过Hadoop的分布式环境处理数据。例如，可以使用HBase的API...