如何不让 Hadoop MapReduce 作业因一个数据库插入失败而失败？答案

【问题标题】：How not to fail Hadoop MapReduce job for one database insert failure?如何不让 Hadoop MapReduce 作业因一个数据库插入失败而失败？
【发布时间】：2015-03-01 17:42:12
【问题描述】：

我正在编写一个 MapReduce 作业来挖掘网络服务器日志。输入来自文本文件，输出到 MySQL 数据库。问题是，如果一条记录由于某种原因无法插入，例如数据超出列大小，则整个作业都会失败，并且不会将任何内容写入数据库。有没有办法让好的记录仍然存在？我想一种方法是验证数据，但这将客户端与数据库模式结合在一起，这对我来说太过分了。 ~~我没有发布代码，因为这不是一个特别的代码问题。~~

编辑：

减速机：

protected void reduce(SkippableLogRecord rec,
        Iterable<NullWritable> values, Context context) {
    String path = rec.getPath().toString();
    path = path.substring(0, min(path.length(), 100));

    try {
        context.write(new DBRecord(rec), NullWritable.get());

        LOGGER.info("Wrote record {}.", path);
    } catch (IOException | InterruptedException e) {
        LOGGER.error("There was a problem when writing out {}.", path, e);
    }
}

日志：

15/03/01 14:35:06 WARN mapred.LocalJobRunner: job_local279539641_0001
java.lang.Exception: java.io.IOException: Data truncation: Data too long for column 'filename' at row 1
    at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:529)
Caused by: java.io.IOException: Data truncation: Data too long for column 'filename' at row 1
    at org.apache.hadoop.mapreduce.lib.db.DBOutputFormat$DBRecordWriter.close(DBOutputFormat.java:103)
    at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.close(ReduceTask.java:550)
    at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:629)
    at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389)
    at org.apache.hadoop.mapred.LocalJobRunner$Job$ReduceTaskRunnable.run(LocalJobRunner.java:319)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
15/03/01 14:35:06 INFO mapred.LocalJobRunner: reduce > reduce
15/03/01 14:35:07 INFO mapreduce.Job: Job job_local279539641_0001 failed with state FAILED due to: NA

【问题讨论】：

嗯。是什么阻止您使用 try catch？
@axiom try-catch 需要围绕引发异常的代码。那个代码不是我的，它是 Hadoop。
这是一个代码问题。代码需要对此类问题进行防御。更改代码以进行尝试/捕获。
显然在你的数据库中。
这是关于效率的，如果你想要高吞吐量，你需要批量插入到一个语句中。如果一个插入失败，它将失败整个批次。您可以相应地调整输出格式/记录写入器。

标签： mysql hadoop mapreduce

【解决方案1】：

回答我自己的问题并查看this SO 帖子，我看到数据库批量写入，在SQLException 上，事务被回滚。所以这解释了我的问题。我想我只需要使数据库列足够大，或者先验证。我也可以创建一个自定义DBOutputFormat/DBRecordWriter，但除非我一次插入一条记录，否则总会有一条错误记录导致整个批次回滚的风险。

public void close(TaskAttemptContext context) throws IOException {
  try {
      LOG.warn("Executing statement:" + statement);   

      statement.executeBatch();
    connection.commit();
  } catch (SQLException e) {
    try {
      connection.rollback();
    }
    catch (SQLException ex) {
      LOG.warn(StringUtils.stringifyException(ex));
    }
    throw new IOException(e.getMessage());
  } finally {
    try {
      statement.close();
      connection.close();
    }
    catch (SQLException ex) {
      throw new IOException(ex.getMessage());
    }
  }
}

【讨论】：