@Test public void testRelativeReadSpeed () throws Exception { int iterations = 1; int slices = 53; HBaseSlicedPyramidIO io = new HBaseSlicedPyramidIO("hadoop-s1", "2181", "hadoop-s1:60000"); TileSerializer<List<Integer>> ks = new KryoSerializer<List<Integer>>(new TypeDescriptor(List.class, new TypeDescriptor(Integer.class))); String table = "nycTaxiHeatmap_sw2015_weekly"; List<TileIndex> indices = Arrays.asList(new TileIndex(9,151,319)); System.out.println("Reading full tile"); TileData<List<Integer>> full = null; long startFull = System.currentTimeMillis(); for (int i=0; i<iterations; ++i) { for (int s=0; s<slices; ++s) { full = io.readTiles(table, ks, indices).get(0); } } long endFull = System.currentTimeMillis(); System.out.println("Checking slice contents"); for (int s=0; s<slices; ++s) { TileData<List<Integer>> slice = io.readTiles(table + "["+s+"]", ks, indices).get(0); for (int x=0; x<full.getDefinition().getXBins(); ++x) { for (int y=0; y<full.getDefinition().getYBins(); ++y) { List<Integer> fullBin = full.getBin(x, y); List<Integer> sliceBin = slice.getBin(x, y); if (null == fullBin || 0 == fullBin.size()) { Assert.assertTrue(null == sliceBin || 0 == sliceBin.size()); } else { Assert.assertEquals(slices, fullBin.size()); Assert.assertEquals(1, sliceBin.size()); Assert.assertEquals(fullBin.get(s), sliceBin.get(0)); } } } } System.out.println("Reading sliced tiles"); long startSlice = System.currentTimeMillis(); for (int i=0; i<iterations; ++i) { for (int s=0; s<slices; ++s) { io.readTiles(table + "["+s+"]", ks, indices); } } long endSlice = System.currentTimeMillis(); System.out.println("Reading unsliced tiles"); indices = Arrays.asList(new TileIndex(0, 0, 0)); TileSerializer<Double> dSerializer = new PrimitiveAvroSerializer<>(Double.class, CodecFactory.bzip2Codec()); String singleSliceTable = "twitter-ebola-p1-heatmap"; long startSingle = System.currentTimeMillis(); for (int i=0; i<iterations; ++i) { for (int s=0; s<slices; ++s) { io.readTiles(singleSliceTable, dSerializer, indices); } } long endSingle = System.currentTimeMillis(); System.out.println("Time for full tile: " + ((endFull - startFull) / 1000.0)); System.out.println("Time for slices: "+((endSlice-startSlice)/1000.0)); System.out.println("Time for unsliced table: "+((endSingle-startSingle)/1000.0)); }
@Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } HBaseSlicedPyramidIO(String zookeeperQuorum, String zookeeperPort, String hbaseMaster); }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } HBaseSlicedPyramidIO(String zookeeperQuorum, String zookeeperPort, String hbaseMaster); @Override HBaseTilePutter getPutter(); void setPyramidding(boolean doPyramidding); static HBaseColumn getSliceColumn(int minSlice, int maxSlice); @Override List<TileData<T>> readTiles(String tableName,
											TileSerializer<T> serializer,
											Iterable<TileIndex> tiles); static List<Pair<Integer, Integer>> decomposeRange(int start, int end); }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } HBaseSlicedPyramidIO(String zookeeperQuorum, String zookeeperPort, String hbaseMaster); @Override HBaseTilePutter getPutter(); void setPyramidding(boolean doPyramidding); static HBaseColumn getSliceColumn(int minSlice, int maxSlice); @Override List<TileData<T>> readTiles(String tableName,
											TileSerializer<T> serializer,
											Iterable<TileIndex> tiles); static List<Pair<Integer, Integer>> decomposeRange(int start, int end);  }