[
https://issues.apache.org/jira/browse/PHOENIX-6944?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17718013#comment-17718013
]
ASF GitHub Bot commented on PHOENIX-6944:
-----------------------------------------
stoty commented on code in PR #1596:
URL: https://github.com/apache/phoenix/pull/1596#discussion_r1181172959
##########
phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputFormat.java:
##########
@@ -87,74 +88,106 @@ public List<InputSplit> getSplits(JobContext context)
throws IOException, Interr
return generateSplits(queryPlan, configuration);
}
- protected List<InputSplit> generateSplits(final QueryPlan qplan,
Configuration config) throws IOException {
+ /**
+ * Randomise the length parameter of the splits to ensure random execution
order.
+ * Yarn orders splits by size before execution.
+ *
+ * @param splits
+ */
+ protected void randomizeSplitLength(List<InputSplit> splits) {
+ LOGGER.info("Randomizing split size");
+ if (splits.size() == 0) {
+ return;
+ }
+ double defaultLength = 1000000d;
+ double totalLength = splits.stream().mapToDouble(s -> {
+ try {
+ return (double)s.getLength();
+ } catch (IOException | InterruptedException e1) {
+ return defaultLength;
+ }
+ }).sum();
+ long avgLength = (long) (totalLength / splits.size());
+ splits.stream().forEach(s -> ((PhoenixInputSplit) s)
+ .setLength(avgLength +
ThreadLocalRandom.current().nextInt(10000)));
+ }
+
+ protected List<InputSplit> generateSplits(final QueryPlan qplan,
Configuration config)
+ throws IOException {
// We must call this in order to initialize the scans and splits from
the query plan
setupParallelScansFromQueryPlan(qplan);
final List<KeyRange> splits = qplan.getSplits();
Preconditions.checkNotNull(splits);
// Get the RegionSizeCalculator
- try(org.apache.hadoop.hbase.client.Connection connection =
-
HBaseFactoryProvider.getHConnectionFactory().createConnection(config)) {
- RegionLocator regionLocator =
connection.getRegionLocator(TableName.valueOf(qplan
- .getTableRef().getTable().getPhysicalName().toString()));
- RegionSizeCalculator sizeCalculator = new
RegionSizeCalculator(regionLocator, connection
- .getAdmin());
+ try (org.apache.hadoop.hbase.client.Connection connection =
+
HBaseFactoryProvider.getHConnectionFactory().createConnection(config)) {
+ RegionLocator regionLocator =
+ connection.getRegionLocator(TableName
+
.valueOf(qplan.getTableRef().getTable().getPhysicalName().toString()));
+ RegionSizeCalculator sizeCalculator =
+ new RegionSizeCalculator(regionLocator,
connection.getAdmin());
- final List<InputSplit> psplits =
Lists.newArrayListWithExpectedSize(splits.size());
- for (List<Scan> scans : qplan.getScans()) {
- // Get the region location
- HRegionLocation location = regionLocator.getRegionLocation(
- scans.get(0).getStartRow(),
- false
- );
+ final List<InputSplit> psplits =
Lists.newArrayListWithExpectedSize(splits.size());
+ for (List<Scan> scans : qplan.getScans()) {
+ // Get the region location
+ HRegionLocation location =
+
regionLocator.getRegionLocation(scans.get(0).getStartRow(), false);
- String regionLocation = location.getHostname();
+ String regionLocation = location.getHostname();
- // Get the region size
- long regionSize = sizeCalculator.getRegionSize(
- location.getRegion().getRegionName()
- );
+ // Get the region size
+ long regionSize =
+
sizeCalculator.getRegionSize(location.getRegion().getRegionName());
- // Generate splits based off statistics, or just region splits?
- boolean splitByStats =
PhoenixConfigurationUtil.getSplitByStats(config);
+ // Generate splits based off statistics, or just region splits?
+ boolean splitByStats =
PhoenixConfigurationUtil.getSplitByStats(config);
- if (splitByStats) {
- for (Scan aScan : scans) {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Split for scan : " + aScan + "with
scanAttribute : " + aScan
- .getAttributesMap() + " [scanCache,
cacheBlock, scanBatch] : [" +
- aScan.getCaching() + ", " +
aScan.getCacheBlocks() + ", " + aScan
- .getBatch() + "] and regionLocation : " +
regionLocation);
- }
+ if (splitByStats) {
+ for (Scan aScan : scans) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Split for scan : " + aScan + "with
scanAttribute : "
+ + aScan.getAttributesMap()
+ + " [scanCache, cacheBlock, scanBatch] :
[" + aScan.getCaching()
+ + ", " + aScan.getCacheBlocks() + ", " +
aScan.getBatch()
+ + "] and regionLocation : " +
regionLocation);
+ }
- psplits.add(new
PhoenixInputSplit(Collections.singletonList(aScan), regionSize,
regionLocation));
- }
+ // The size is bogus, but it's not a problem
+ psplits.add(new
PhoenixInputSplit(Collections.singletonList(aScan),
+ regionSize, regionLocation));
+ }
} else {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Scan count[" + scans.size() + "] : " +
Bytes.toStringBinary(scans
- .get(0).getStartRow()) + " ~ " +
Bytes.toStringBinary(scans.get(scans
- .size() - 1).getStopRow()));
- LOGGER.debug("First scan : " + scans.get(0) + "with
scanAttribute : " + scans
- .get(0).getAttributesMap() + " [scanCache,
cacheBlock, scanBatch] : " +
- "[" + scans.get(0).getCaching() + ", " +
scans.get(0).getCacheBlocks()
- + ", " + scans.get(0).getBatch() + "] and
regionLocation : " +
- regionLocation);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Scan count[" + scans.size() + "] : "
+ +
Bytes.toStringBinary(scans.get(0).getStartRow()) + " ~ "
+ + Bytes.toStringBinary(scans.get(scans.size()
- 1).getStopRow()));
+ LOGGER.debug("First scan : " + scans.get(0) + "with
scanAttribute : "
+ + scans.get(0).getAttributesMap()
+ + " [scanCache, cacheBlock, scanBatch] : " +
"["
+ + scans.get(0).getCaching() + ", " +
scans.get(0).getCacheBlocks()
+ + ", " + scans.get(0).getBatch() + "] and
regionLocation : "
+ + regionLocation);
- for (int i = 0, limit = scans.size(); i < limit; i++) {
- LOGGER.debug("EXPECTED_UPPER_REGION_KEY[" + i + "] : "
+ Bytes
- .toStringBinary(scans.get(i).getAttribute
-
(BaseScannerRegionObserver.EXPECTED_UPPER_REGION_KEY)));
+ for (int i = 0, limit = scans.size(); i < limit; i++) {
+ LOGGER.debug("EXPECTED_UPPER_REGION_KEY[" + i + "]
: "
+ +
Bytes.toStringBinary(scans.get(i).getAttribute(
+
BaseScannerRegionObserver.EXPECTED_UPPER_REGION_KEY)));
+ }
}
+
+ psplits.add(new PhoenixInputSplit(scans, regionSize,
regionLocation));
}
+ }
- psplits.add(new PhoenixInputSplit(scans, regionSize,
regionLocation));
+ if
(PhoenixConfigurationUtil.isMRRandomizeMapperExecutionOrder(config)) {
Review Comment:
This is the only code change, the rest of the changes in thsis method are
just reformats
##########
phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputFormat.java:
##########
@@ -87,74 +88,106 @@ public List<InputSplit> getSplits(JobContext context)
throws IOException, Interr
return generateSplits(queryPlan, configuration);
}
- protected List<InputSplit> generateSplits(final QueryPlan qplan,
Configuration config) throws IOException {
+ /**
+ * Randomise the length parameter of the splits to ensure random execution
order.
+ * Yarn orders splits by size before execution.
+ *
+ * @param splits
+ */
+ protected void randomizeSplitLength(List<InputSplit> splits) {
+ LOGGER.info("Randomizing split size");
+ if (splits.size() == 0) {
+ return;
+ }
+ double defaultLength = 1000000d;
+ double totalLength = splits.stream().mapToDouble(s -> {
+ try {
+ return (double)s.getLength();
+ } catch (IOException | InterruptedException e1) {
+ return defaultLength;
+ }
+ }).sum();
+ long avgLength = (long) (totalLength / splits.size());
+ splits.stream().forEach(s -> ((PhoenixInputSplit) s)
+ .setLength(avgLength +
ThreadLocalRandom.current().nextInt(10000)));
+ }
+
+ protected List<InputSplit> generateSplits(final QueryPlan qplan,
Configuration config)
+ throws IOException {
// We must call this in order to initialize the scans and splits from
the query plan
setupParallelScansFromQueryPlan(qplan);
final List<KeyRange> splits = qplan.getSplits();
Preconditions.checkNotNull(splits);
// Get the RegionSizeCalculator
- try(org.apache.hadoop.hbase.client.Connection connection =
-
HBaseFactoryProvider.getHConnectionFactory().createConnection(config)) {
- RegionLocator regionLocator =
connection.getRegionLocator(TableName.valueOf(qplan
- .getTableRef().getTable().getPhysicalName().toString()));
- RegionSizeCalculator sizeCalculator = new
RegionSizeCalculator(regionLocator, connection
- .getAdmin());
+ try (org.apache.hadoop.hbase.client.Connection connection =
+
HBaseFactoryProvider.getHConnectionFactory().createConnection(config)) {
+ RegionLocator regionLocator =
+ connection.getRegionLocator(TableName
+
.valueOf(qplan.getTableRef().getTable().getPhysicalName().toString()));
+ RegionSizeCalculator sizeCalculator =
+ new RegionSizeCalculator(regionLocator,
connection.getAdmin());
- final List<InputSplit> psplits =
Lists.newArrayListWithExpectedSize(splits.size());
- for (List<Scan> scans : qplan.getScans()) {
- // Get the region location
- HRegionLocation location = regionLocator.getRegionLocation(
- scans.get(0).getStartRow(),
- false
- );
+ final List<InputSplit> psplits =
Lists.newArrayListWithExpectedSize(splits.size());
+ for (List<Scan> scans : qplan.getScans()) {
+ // Get the region location
+ HRegionLocation location =
+
regionLocator.getRegionLocation(scans.get(0).getStartRow(), false);
- String regionLocation = location.getHostname();
+ String regionLocation = location.getHostname();
- // Get the region size
- long regionSize = sizeCalculator.getRegionSize(
- location.getRegion().getRegionName()
- );
+ // Get the region size
+ long regionSize =
+
sizeCalculator.getRegionSize(location.getRegion().getRegionName());
- // Generate splits based off statistics, or just region splits?
- boolean splitByStats =
PhoenixConfigurationUtil.getSplitByStats(config);
+ // Generate splits based off statistics, or just region splits?
+ boolean splitByStats =
PhoenixConfigurationUtil.getSplitByStats(config);
- if (splitByStats) {
- for (Scan aScan : scans) {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Split for scan : " + aScan + "with
scanAttribute : " + aScan
- .getAttributesMap() + " [scanCache,
cacheBlock, scanBatch] : [" +
- aScan.getCaching() + ", " +
aScan.getCacheBlocks() + ", " + aScan
- .getBatch() + "] and regionLocation : " +
regionLocation);
- }
+ if (splitByStats) {
+ for (Scan aScan : scans) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Split for scan : " + aScan + "with
scanAttribute : "
+ + aScan.getAttributesMap()
+ + " [scanCache, cacheBlock, scanBatch] :
[" + aScan.getCaching()
+ + ", " + aScan.getCacheBlocks() + ", " +
aScan.getBatch()
+ + "] and regionLocation : " +
regionLocation);
+ }
- psplits.add(new
PhoenixInputSplit(Collections.singletonList(aScan), regionSize,
regionLocation));
- }
+ // The size is bogus, but it's not a problem
+ psplits.add(new
PhoenixInputSplit(Collections.singletonList(aScan),
+ regionSize, regionLocation));
+ }
} else {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Scan count[" + scans.size() + "] : " +
Bytes.toStringBinary(scans
- .get(0).getStartRow()) + " ~ " +
Bytes.toStringBinary(scans.get(scans
- .size() - 1).getStopRow()));
- LOGGER.debug("First scan : " + scans.get(0) + "with
scanAttribute : " + scans
- .get(0).getAttributesMap() + " [scanCache,
cacheBlock, scanBatch] : " +
- "[" + scans.get(0).getCaching() + ", " +
scans.get(0).getCacheBlocks()
- + ", " + scans.get(0).getBatch() + "] and
regionLocation : " +
- regionLocation);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Scan count[" + scans.size() + "] : "
+ +
Bytes.toStringBinary(scans.get(0).getStartRow()) + " ~ "
+ + Bytes.toStringBinary(scans.get(scans.size()
- 1).getStopRow()));
+ LOGGER.debug("First scan : " + scans.get(0) + "with
scanAttribute : "
+ + scans.get(0).getAttributesMap()
+ + " [scanCache, cacheBlock, scanBatch] : " +
"["
+ + scans.get(0).getCaching() + ", " +
scans.get(0).getCacheBlocks()
+ + ", " + scans.get(0).getBatch() + "] and
regionLocation : "
+ + regionLocation);
- for (int i = 0, limit = scans.size(); i < limit; i++) {
- LOGGER.debug("EXPECTED_UPPER_REGION_KEY[" + i + "] : "
+ Bytes
- .toStringBinary(scans.get(i).getAttribute
-
(BaseScannerRegionObserver.EXPECTED_UPPER_REGION_KEY)));
+ for (int i = 0, limit = scans.size(); i < limit; i++) {
+ LOGGER.debug("EXPECTED_UPPER_REGION_KEY[" + i + "]
: "
+ +
Bytes.toStringBinary(scans.get(i).getAttribute(
+
BaseScannerRegionObserver.EXPECTED_UPPER_REGION_KEY)));
+ }
}
+
+ psplits.add(new PhoenixInputSplit(scans, regionSize,
regionLocation));
}
+ }
- psplits.add(new PhoenixInputSplit(scans, regionSize,
regionLocation));
+ if
(PhoenixConfigurationUtil.isMRRandomizeMapperExecutionOrder(config)) {
Review Comment:
This is the only code change, the rest of the changes in this method are
just reformats
> Randomize mapper task ordering for MR tools
> -------------------------------------------
>
> Key: PHOENIX-6944
> URL: https://issues.apache.org/jira/browse/PHOENIX-6944
> Project: Phoenix
> Issue Type: Improvement
> Components: core
> Reporter: Istvan Toth
> Priority: Major
> Labels: perf
>
> Currently, splits are generated by PhoenixInputFormat are in ascending order.
> MR does not use this ordering directly, it instead orders the partitions by
> size in descending order.
> We set the sizes of the splits to the region size. (Even when splitting by
> guideposts, but this not really a problem)
> The result is that mapper jobs are grouped by regions, so usually all the
> mappers running are working on one, or few regions. As a result we have the
> following problems:
> Read hotspotting:
> All scan operations for the indexing job hit the same one or few region
> servers, causing high loads and slowdowns.
> Write hotspotting:
> If the data rowkeys and index rowkeys strongly correlate, then the data read
> from one or few data regions will be written to one or few index regions,
> causing high loads and slowdowns. This is a bit of a corner case, we have
> obeserved it when building an index for a column which starts with the same
> bytes as the primary key for the data table.
> We can improve this by making sure that the generate mapper jobs are executed
> in a random order. The only way to change the execution order is to
> manipulate the length of the splits. As the length is only used for ordering,
> and calculating completion percentage, this is unlikely to cause problems (we
> already specify wildly off lengths when splitting by guidepost )
> I've run some test on a 50M row, 40GB data table, generating secondary
> indexes for a correlated field and for a random field:
> The test system has three RS workers, and 12 yarn slots for running IndexTool
> ||Index rebuild time||on correlated field||on random field||
> |w/o randomization|50 min|28 min|
> |w/ randomization|30 min|23 min|
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
