[
https://issues.apache.org/jira/browse/BEAM-4783?focusedWorklogId=160654&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-160654
]
ASF GitHub Bot logged work on BEAM-4783:
----------------------------------------
Author: ASF GitHub Bot
Created on: 30/Oct/18 14:34
Start Date: 30/Oct/18 14:34
Worklog Time Spent: 10m
Work Description: timrobertson100 closed pull request #6884: [BEAM-4783]
Fix invalid parameter to set the partitioner in Spark GbK
URL: https://github.com/apache/beam/pull/6884
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/io/SourceRDD.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/io/SourceRDD.java
index dd3137d0af8..304f3f46c8e 100644
---
a/runners/spark/src/main/java/org/apache/beam/runners/spark/io/SourceRDD.java
+++
b/runners/spark/src/main/java/org/apache/beam/runners/spark/io/SourceRDD.java
@@ -99,11 +99,8 @@ public Bounded(
@Override
public Partition[] getPartitions() {
try {
- List<? extends Source<T>> partitionedSources;
- if (bundleSize > 0) {
- partitionedSources = source.split(bundleSize, options.get());
- } else {
- long desiredSizeBytes = DEFAULT_BUNDLE_SIZE;
+ long desiredSizeBytes = (bundleSize > 0) ? bundleSize :
DEFAULT_BUNDLE_SIZE;
+ if (bundleSize == 0) {
try {
desiredSizeBytes = source.getEstimatedSizeBytes(options.get()) /
numPartitions;
} catch (Exception e) {
@@ -113,8 +110,10 @@ public Bounded(
source,
DEFAULT_BUNDLE_SIZE);
}
- partitionedSources = source.split(desiredSizeBytes, options.get());
}
+
+ List<? extends Source<T>> partitionedSources =
+ source.split(desiredSizeBytes, options.get());
Partition[] partitions = new
SourcePartition[partitionedSources.size()];
for (int i = 0; i < partitionedSources.size(); i++) {
partitions[i] = new SourcePartition<>(id(), i,
partitionedSources.get(i));
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/GroupCombineFunctions.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/GroupCombineFunctions.java
index 8216b57980f..f79dac2e9c1 100644
---
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/GroupCombineFunctions.java
+++
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/GroupCombineFunctions.java
@@ -18,6 +18,7 @@
package org.apache.beam.runners.spark.translation;
import com.google.common.base.Optional;
+import javax.annotation.Nullable;
import org.apache.beam.runners.spark.coders.CoderHelpers;
import org.apache.beam.runners.spark.util.ByteArray;
import org.apache.beam.sdk.coders.Coder;
@@ -28,7 +29,7 @@
import org.apache.beam.sdk.util.WindowedValue.WindowedValueCoder;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.WindowingStrategy;
-import org.apache.spark.HashPartitioner;
+import org.apache.spark.Partitioner;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
@@ -45,7 +46,7 @@
JavaRDD<WindowedValue<KV<K, V>>> rdd,
Coder<K> keyCoder,
WindowedValueCoder<V> wvCoder,
- boolean defaultParallelism) {
+ @Nullable Partitioner partitioner) {
// we use coders to convert objects in the PCollection to byte arrays, so
they
// can be transferred over the network for the shuffle.
JavaPairRDD<ByteArray, byte[]> pairRDD =
@@ -53,13 +54,10 @@
.map(WindowingHelpers.unwindowFunction())
.mapToPair(TranslationUtils.toPairFunction())
.mapToPair(CoderHelpers.toByteFunction(keyCoder, wvCoder));
- JavaPairRDD<ByteArray, Iterable<byte[]>> groupedRDD;
- if (defaultParallelism) {
- groupedRDD =
- pairRDD.groupByKey(new
HashPartitioner(rdd.rdd().sparkContext().defaultParallelism()));
- } else {
- groupedRDD = pairRDD.groupByKey();
- }
+
+ // If no partitioner is passed, the default group by key operation is
called
+ JavaPairRDD<ByteArray, Iterable<byte[]>> groupedRDD =
+ (partitioner != null) ? pairRDD.groupByKey(partitioner) :
pairRDD.groupByKey();
// using mapPartitions allows to preserve the partitioner
// and avoid unnecessary shuffle downstream.
@@ -93,10 +91,10 @@
// can be transferred over the network for the shuffle.
// for readability, we add comments with actual type next to byte[].
// to shorten line length, we use:
- //---- WV: WindowedValue
- //---- Iterable: Itr
- //---- AccumT: A
- //---- InputT: I
+ // ---- WV: WindowedValue
+ // ---- Iterable: Itr
+ // ---- AccumT: A
+ // ---- InputT: I
JavaRDD<byte[]> inputRDDBytes =
rdd.map(CoderHelpers.toByteFunction(wviCoder));
if (inputRDDBytes.isEmpty()) {
@@ -173,10 +171,10 @@
// can be transferred over the network for the shuffle.
// for readability, we add comments with actual type next to byte[].
// to shorten line length, we use:
- //---- WV: WindowedValue
- //---- Iterable: Itr
- //---- AccumT: A
- //---- InputT: I
+ // ---- WV: WindowedValue
+ // ---- Iterable: Itr
+ // ---- AccumT: A
+ // ---- InputT: I
JavaPairRDD<ByteArray, byte[]> inRddDuplicatedKeyPairBytes =
inRddDuplicatedKeyPair.mapToPair(CoderHelpers.toByteFunction(keyCoder,
wkviCoder));
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/TransformTranslator.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/TransformTranslator.java
index abe93eb0fe4..4c4851a1a56 100644
---
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/TransformTranslator.java
+++
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/TransformTranslator.java
@@ -65,6 +65,8 @@
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.spark.Accumulator;
+import org.apache.spark.HashPartitioner;
+import org.apache.spark.Partitioner;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
@@ -130,17 +132,14 @@ public void evaluate(GroupByKey<K, V> transform,
EvaluationContext context) {
WindowedValue.FullWindowedValueCoder.of(coder.getValueCoder(),
windowFn.windowCoder());
// --- group by key only.
+ Long bundleSize =
+
context.getSerializableOptions().get().as(SparkPipelineOptions.class).getBundleSize();
+ Partitioner partitioner =
+ (bundleSize > 0)
+ ? new
HashPartitioner(context.getSparkContext().defaultParallelism())
+ : null;
JavaRDD<WindowedValue<KV<K, Iterable<WindowedValue<V>>>>> groupedByKey
=
- GroupCombineFunctions.groupByKeyOnly(
- inRDD,
- keyCoder,
- wvCoder,
- context
- .getSerializableOptions()
- .get()
- .as(SparkPipelineOptions.class)
- .getBundleSize()
- > 0);
+ GroupCombineFunctions.groupByKeyOnly(inRDD, keyCoder, wvCoder,
partitioner);
// --- now group also by window.
// for batch, GroupAlsoByWindow uses an in-memory StateInternals.
@@ -433,7 +432,7 @@ public String toNativeString() {
WindowedValue.FullWindowedValueCoder.of(kvCoder.getValueCoder(),
windowCoder);
JavaRDD<WindowedValue<KV<K, Iterable<WindowedValue<V>>>>> groupRDD =
- GroupCombineFunctions.groupByKeyOnly(kvInRDD, keyCoder, wvCoder, true);
+ GroupCombineFunctions.groupByKeyOnly(kvInRDD, keyCoder, wvCoder, null);
return groupRDD
.map(
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
index 8e54ebf7bfe..f20eda5f5b6 100644
---
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
+++
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
@@ -301,7 +301,7 @@ public void evaluate(GroupByKey<K, V> transform,
EvaluationContext context) {
JavaDStream<WindowedValue<KV<K, Iterable<WindowedValue<V>>>>>
groupedByKeyStream =
dStream.transform(
rdd ->
- GroupCombineFunctions.groupByKeyOnly(rdd,
coder.getKeyCoder(), wvCoder, true));
+ GroupCombineFunctions.groupByKeyOnly(rdd,
coder.getKeyCoder(), wvCoder, null));
// --- now group also by window.
JavaDStream<WindowedValue<KV<K, Iterable<V>>>> outStream =
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 160654)
Time Spent: 5h 50m (was: 5h 40m)
> Add bundleSize parameter to control splitting of Spark sources (useful for
> Dynamic Allocation)
> ----------------------------------------------------------------------------------------------
>
> Key: BEAM-4783
> URL: https://issues.apache.org/jira/browse/BEAM-4783
> Project: Beam
> Issue Type: Improvement
> Components: runner-spark
> Affects Versions: 2.8.0
> Reporter: Kyle Winkelman
> Assignee: Kyle Winkelman
> Priority: Major
> Fix For: 2.8.0, 2.9.0
>
> Time Spent: 5h 50m
> Remaining Estimate: 0h
>
> When the spark-runner is used along with the configuration
> spark.dynamicAllocation.enabled=true the SourceRDD does not detect this. It
> then falls back to the value calculated in this description:
> // when running on YARN/SparkDeploy it's the result of max(totalCores,
> 2).
> // when running on Mesos it's 8.
> // when running local it's the total number of cores (local = 1,
> local[N] = N,
> // local[*] = estimation of the machine's cores).
> // ** the configuration "spark.default.parallelism" takes precedence
> over all of the above **
> So in most cases this default is quite small. This is an issue when using a
> very large input file as it will only get split in half.
> I believe that when Dynamic Allocation is enable the SourceRDD should use the
> DEFAULT_BUNDLE_SIZE and possibly expose a SparkPipelineOptions that allows
> you to change this DEFAULT_BUNDLE_SIZE.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
