如何在 MapReduce 中使用 sqoop 生成的类？答案

【问题标题】：How to use the sqoop generated class in MapReduce?如何在 MapReduce 中使用 sqoop 生成的类？
【发布时间】：2012-07-18 21:19:12
【问题描述】：

一个 sqoop 查询生成一个 java 文件，其中包含一个类，该类包含在 mapreduce 中访问每行的列数据的代码。（Sqoop 导入是在没有 --as-sequencefile 选项的文本中完成的，每条记录有 1 行，列之间有逗号）但是我们如何实际使用它呢？

我在这个类中找到了一个公共方法 parse()，它将 Text 作为输入并填充类的所有成员，所以为了练习，我修改了 wordcount 应用程序，将映射器中 TextInputFormat 中的一行文本转换为sqoop 生成的类的实例。但是当我调用 parse() 方法时，这会导致“未报告的 exception.com.cloudera.sqoop.lib.RecordParser.ParseError; 必须被捕获或声明为抛出”。

可以通过这种方式完成，还是使用每个记录中的数据填充类所必需的自定义 InputFormat ？

【问题讨论】：

标签： class import mapreduce sqoop

【解决方案1】：

好吧，一旦你发现这似乎很明显，但作为一个 java 初学者，这可能需要一些时间。

首先配置你的项目：只需在源文件夹中添加 sqoop 生成的 .java 文件。我使用 eclipse 将它导入到我的类源文件夹中。

然后确保您正确配置了项目的 java 构建路径：

在项目的properties/java build path/libraries/add external jar中添加如下jar文件：（对于 hadoop cdh4+）：

/usr/lib/hadoop/hadoop-common.jar 
/usr/lib/hadoop-[version]-mapreduce/hadoop-core.jar
/usr/lib/sqoop/sqoop-[sqoop-version]-cdh[cdh-version].jar

然后调整您的 mapreduce 源代码：先配置一下：

public int run(String [] args) throws exception
{
 Job job = new Job(getConf());
 job.setJarByClass(YourClass.class);
 job.setMapperClass(SqoopImportMap.class);
 job.setReducerClass(SqoopImprtReduce.class);

 FileInputFormat.addInputPath((job,"hdfs_path_to_your_sqoop_imported_file"));
 FileOutputFormat.setOutputPath((job,"hdfs_output_path"));

 // I simply use text as output for the mapper but it can be any class you designed
 // as long as you implement it as a Writable
 job.setMapOutputKeyClass(Text.Class);
 job.setMapOutputValueClass(Text.Class);

 job.setOutputKeyClass(Text.Class);
 job.setOutputValueClass(Text.Class);
 ...

现在配置您的映射器类。假设您的 sqoop 导入的 java 文件名为 Sqimp.java：并且您导入的表具有以下列：id、name、age 您的映射器类应如下所示：

 public static class SqoopImportMap
 extends Mapper<LongWritable, Text, Text, Text> 
 {

 public void map(LongWritable k, Text v, Context context)
 {
  Sqimp s = new Sqimp(); 
  try
  {
  // this is where the code generated by sqoop is used.
  // it automatically casts one line of the imported data into an instance of the generated class, 
  // to let you access the data inside the columns easily
   s.parse(v);
  } 
  catch(ParseError pe) {// do something if there is an error.}

  try
  {
   // now the imported data is accessible:
   // e.g
   if (s.age>30)
   {
    // submit the selected data to the mapper's output as a key value pair.
    context.write(new Text(s.age),new Text(s.id));
   }
  }
  catch(Exception ex)
  {//do something about the error}
 }
}

【讨论】：