Hadoopカウンターとデータクリーニングの適用

パッケージ com.atguigu.mapreduce.weblog;
java.io.IOException をインポートします。
org.apache.hadoop.io.LongWritable をインポートします。
org.apache.hadoop.io.NullWritable をインポートします。
org.apache.hadoop.io.Text をインポートします。
org.apache.hadoop.mapreduce.Mapper をインポートします。
パブリッククラス LogMapper は Mapper<LongWritable, Text, Text, NullWritable> を拡張します{
  テキスト k = 新しいテキスト();
  @オーバーライド
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
   // 1 1行のデータを取得します。String line = value.toString();
   // 2 ログを解析する boolean result = parseLog(line,context);
   // 3 ログが不正なので終了 if (!result) {
     戻る;
   }
   // 4 キーを設定する
   k.set(行);
   // 5 データを書き込む context.write(k, NullWritable.get());
  }
  // 2 ログを解析する private boolean parseLog(String line, Context context) {
   // 1 インターセプト String[] fields = line.split(" ");
   // 長さが11より大きい2つのログは有効です if (fields.length > 11) {
     // システムカウンター context.getCounter("map", "true").increment(1);
     true を返します。
   }それ以外 {
     context.getCounter("map", "false").increment(1);
     false を返します。
   }
  }
}

パッケージ com.atguigu.mapreduce.weblog;
org.apache.hadoop.conf.Configuration をインポートします。
org.apache.hadoop.fs.Path をインポートします。
org.apache.hadoop.io.NullWritable をインポートします。
org.apache.hadoop.io.Text をインポートします。
org.apache.hadoop.mapreduce.Job をインポートします。
org.apache.hadoop.mapreduce.lib.input.FileInputFormat をインポートします。
org.apache.hadoop.mapreduce.lib.output.FileOutputFormat をインポートします。
パブリッククラスLogDriver {
  パブリック静的void main(String[] args)は例外をスローします{
// 入力パスと出力パスは、コンピューターの実際の入力パスと出力パスに従って設定する必要があります args = new String[] { "e:/input/inputlog", "e:/output1" };
   // 1 ジョブ情報を取得する Configuration conf = new Configuration();
   ジョブ job = Job.getInstance(conf);
   // 2 jar パッケージをロードします job.setJarByClass(LogDriver.class);
   // 3 つの関連マップ
   ジョブを LogMapper.class に設定します。
   // 4 最終出力タイプを設定する job.setOutputKeyClass(Text.class);
   ジョブの出力値クラスを設定します(NullWritable.class)。
   // 削減タスクの数を0に設定する
   ジョブのタスク数を減らす(0)
   // 5 入力パスと出力パスを設定する FileInputFormat.setInputPaths(job, new Path(args[0]));
   FileOutputFormat.setOutputPath(ジョブ、新しいパス(args[1]));
   // 6 送信 job.waitForCompletion(true);
  }
}