This closes apache#2169

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java

@@ -32,6 +32,7 @@
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.transforms.Aggregator;
 import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.util.IOChannelUtils;
 import org.apache.spark.Accumulator;
 
 /**
@@ -40,12 +41,10 @@
  */
 public class SparkRuntimeContext implements Serializable {
  private final String serializedPipelineOptions;
+ private transient CoderRegistry coderRegistry;
 
- /**
- * Map fo names to Beam aggregators.
- */
+ // map for names to Beam aggregators.
  private final Map<String, Aggregator<?, ?>> aggregators = new HashMap<>();
- private transient CoderRegistry coderRegistry;
 
  SparkRuntimeContext(Pipeline pipeline) {
  this.serializedPipelineOptions = serializePipelineOptions(pipeline.getOptions());
@@ -67,8 +66,8 @@ private static PipelineOptions deserializePipelineOptions(String serializedPipel
  }
  }
 
- public synchronized PipelineOptions getPipelineOptions() {
- return deserializePipelineOptions(serializedPipelineOptions);
+ public PipelineOptions getPipelineOptions() {
+ return PipelineOptionsHolder.getOrInit(serializedPipelineOptions);
  }
 
  /**
@@ -118,6 +117,24 @@ public CoderRegistry getCoderRegistry() {
  return coderRegistry;
  }
 
+ private static class PipelineOptionsHolder {
+ // on executors, this should deserialize once.
+ private static transient volatile PipelineOptions pipelineOptions = null;
+
+ static PipelineOptions getOrInit(String serializedPipelineOptions) {
+ if (pipelineOptions == null) {
+ synchronized (PipelineOptionsHolder.class) {
+ if (pipelineOptions == null) {
+ pipelineOptions = deserializePipelineOptions(serializedPipelineOptions);
+ }
+ }
+ // register IO factories.
+ IOChannelUtils.registerIOFactoriesAllowOverride(pipelineOptions);
+ }
+ return pipelineOptions;
+ }
+ }
+
  /**
  * Initialize spark aggregators exactly once.
  *