<commit_before>/*
Copyright 2012 Urban Airship and Contributors
*/

package com.urbanairship.datacube.backfill;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import com.google.common.io.Closeables;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.fs.Path;

import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.urbanairship.datacube.Deserializer;
import com.urbanairship.datacube.collectioninputformat.CollectionInputFormat;

/**
 * Backfilling is a 3 step process:
 *  (1) snapshot the live cube data into a so-called snapshot table
 *  (2) do the application-level event processing into a cube that is stored in some non-visible table
 *  (3) Merge the snapshot, the backfill table, and the live cube, making changes to the live cube.
 *  
 *  This MR job does step 3.
 *  
 *  TODO InputSplit locality for mapreduce
 */
public class HBaseBackfillMerger implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(HBaseBackfillMerger.class);
    
    static final String CONFKEY_COLUMN_FAMILY = "hbasebackfiller.cf";
    static final String CONFKEY_LIVECUBE_TABLE_NAME = "hbasebackfiller.liveCubeTableName";
    static final String CONFKEY_SNAPSHOT_TABLE_NAME = "hbasebackfiller.snapshotTableName";
    static final String CONFKEY_BACKFILLED_TABLE_NAME = "hbasebackfiller.backfilledTableName";
    static final String CONFKEY_DESERIALIZER = "hbasebackfiller.deserializerClassName";
    
    private final Configuration conf;
    private final byte[] cubeNameKeyPrefix; 
    private final byte[] liveCubeTableName;
    private final byte[] snapshotTableName;
    private final byte[] backfilledTableName;
    private final byte[] cf;
    private final Class<? extends Deserializer<?>> opDeserializer;
    
    private static final byte ff = (byte)0xFF;
    private static final byte[] fiftyBytesFF = new byte[] {ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff,
            ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff,
            ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff};
    
    public HBaseBackfillMerger(Configuration conf, byte[] cubeNameKeyPrefix, byte[] liveCubeTableName, 
            byte[] snapshotTableName, byte[] backfilledTableName, byte[] cf, 
            Class<? extends Deserializer<?>> opDeserializer) {
        this.conf = conf;
        this.cubeNameKeyPrefix = cubeNameKeyPrefix;
        this.liveCubeTableName = liveCubeTableName;
        this.snapshotTableName = snapshotTableName;
        this.backfilledTableName = backfilledTableName;
        this.cf = cf;
        this.opDeserializer = opDeserializer;
    }

    /**
     * A wrapper around {@link #runWithCheckedExceptions()} that rethrows IOExceptions as
     * RuntimeExceptions.
     */
    @Override
    public void run() {
        try {
            this.runWithCheckedExceptions();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    
    public boolean runWithCheckedExceptions() throws IOException, InterruptedException {
        HTable backfilledHTable = null;
        HTable liveCubeHTable = null;
        ResultScanner liveCubeScanner = null;
        try {
            // If the live table is empty, we just efficiently copy the backfill in
            liveCubeHTable = new HTable(conf, liveCubeTableName);
            liveCubeScanner = liveCubeHTable.getScanner(cf);
            if (!liveCubeScanner.iterator().hasNext()) {
                log.info("Live cube is empty, running a straight copy from the backfill table");
                
                HBaseSnapshotter hbaseSnapshotter = new HBaseSnapshotter(conf, backfilledTableName, cf, liveCubeTableName, 
                  new Path("/tmp/backfill_snapshot_hfiles"), true, null, null);
                return hbaseSnapshotter.runWithCheckedExceptions();
            }  else {          
          
                Job job = new Job(conf);
                backfilledHTable = new HTable(conf, backfilledTableName);
                
                Pair<byte[][],byte[][]> allRegionsStartAndEndKeys = backfilledHTable.getStartEndKeys();
                byte[][] internalSplitKeys = BackfillUtil.getSplitKeys(allRegionsStartAndEndKeys);
                Collection<Scan> scans = scansThisCubeOnly(cubeNameKeyPrefix, internalSplitKeys);
                
                // Get the scans that will cover this table, and store them in the job configuration to be used
                // as input splits.
                
                if(log.isDebugEnabled()) {
                    log.debug("Scans: " + scans);
                }
                CollectionInputFormat.setCollection(job, Scan.class, scans);
                
                // We cannot allow map tasks to retry, or we could increment the same key multiple times.
                job.getConfiguration().set("mapred.map.max.attempts", "1");<commit_msg>Close ResultScanner as soon as possible in HBaseBackfillMerger<commit_after>/*
Copyright 2012 Urban Airship and Contributors
*/

package com.urbanairship.datacube.backfill;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import com.google.common.io.Closeables;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.fs.Path;

import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.urbanairship.datacube.Deserializer;
import com.urbanairship.datacube.collectioninputformat.CollectionInputFormat;

/**
 * Backfilling is a 3 step process:
 *  (1) snapshot the live cube data into a so-called snapshot table
 *  (2) do the application-level event processing into a cube that is stored in some non-visible table
 *  (3) Merge the snapshot, the backfill table, and the live cube, making changes to the live cube.
 *  
 *  This MR job does step 3.
 *  
 *  TODO InputSplit locality for mapreduce
 */
public class HBaseBackfillMerger implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(HBaseBackfillMerger.class);
    
    static final String CONFKEY_COLUMN_FAMILY = "hbasebackfiller.cf";
    static final String CONFKEY_LIVECUBE_TABLE_NAME = "hbasebackfiller.liveCubeTableName";
    static final String CONFKEY_SNAPSHOT_TABLE_NAME = "hbasebackfiller.snapshotTableName";
    static final String CONFKEY_BACKFILLED_TABLE_NAME = "hbasebackfiller.backfilledTableName";
    static final String CONFKEY_DESERIALIZER = "hbasebackfiller.deserializerClassName";
    
    private final Configuration conf;
    private final byte[] cubeNameKeyPrefix; 
    private final byte[] liveCubeTableName;
    private final byte[] snapshotTableName;
    private final byte[] backfilledTableName;
    private final byte[] cf;
    private final Class<? extends Deserializer<?>> opDeserializer;
    
    private static final byte ff = (byte)0xFF;
    private static final byte[] fiftyBytesFF = new byte[] {ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff,
            ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff,
            ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff, ff};
    
    public HBaseBackfillMerger(Configuration conf, byte[] cubeNameKeyPrefix, byte[] liveCubeTableName, 
            byte[] snapshotTableName, byte[] backfilledTableName, byte[] cf, 
            Class<? extends Deserializer<?>> opDeserializer) {
        this.conf = conf;
        this.cubeNameKeyPrefix = cubeNameKeyPrefix;
        this.liveCubeTableName = liveCubeTableName;
        this.snapshotTableName = snapshotTableName;
        this.backfilledTableName = backfilledTableName;
        this.cf = cf;
        this.opDeserializer = opDeserializer;
    }

    /**
     * A wrapper around {@link #runWithCheckedExceptions()} that rethrows IOExceptions as
     * RuntimeExceptions.
     */
    @Override
    public void run() {
        try {
            this.runWithCheckedExceptions();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    
    public boolean runWithCheckedExceptions() throws IOException, InterruptedException {
        HTable backfilledHTable = null;
        HTable liveCubeHTable = null;
        ResultScanner liveCubeScanner = null;
        try {
            // If the live table is empty, we just efficiently copy the backfill in
            liveCubeHTable = new HTable(conf, liveCubeTableName);
            liveCubeScanner = liveCubeHTable.getScanner(cf);
            boolean liveCubeIsEmpty = !liveCubeScanner.iterator().hasNext();
            liveCubeScanner.close();
            if (liveCubeIsEmpty) {
                log.info("Live cube is empty, running a straight copy from the backfill table");
                
                HBaseSnapshotter hbaseSnapshotter = new HBaseSnapshotter(conf, backfilledTableName, cf, liveCubeTableName, 
                  new Path("/tmp/backfill_snapshot_hfiles"), true, null, null);
                return hbaseSnapshotter.runWithCheckedExceptions();
            }  else {          
          
                Job job = new Job(conf);
                backfilledHTable = new HTable(conf, backfilledTableName);
                
                Pair<byte[][],byte[][]> allRegionsStartAndEndKeys = backfilledHTable.getStartEndKeys();
                byte[][] internalSplitKeys = BackfillUtil.getSplitKeys(allRegionsStartAndEndKeys);
                Collection<Scan> scans = scansThisCubeOnly(cubeNameKeyPrefix, internalSplitKeys);
                
                // Get the scans that will cover this table, and store them in the job configuration to be used
                // as input splits.
                
                if(log.isDebugEnabled()) {
                    log.debug("Scans: " + scans);
                }
                CollectionInputFormat.setCollection(job, Scan.class, scans);
                
                // We cannot allow map tasks to retry, or we could increment the same key multiple times.
                job.getConfiguration().set("mapred.map.max.attempts", "1");
{'fraction_non_alphanumeric': 0.08795657374183458, 'fraction_numerical': 0.002392124390468304, 'mean_word_length': 2.8437057991513437, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2563417', 'n_tokens_mistral': 3243, 'n_tokens_neox': 3054, 'n_words': 846}