apache / apache/beam

[Bug]: Side Input Singleton View throw error when impulse period is short: PCollection with more than one element accessed as a singleton view

Open
#26,465 11 comments 0 reactions 0 assignees View on GitHub
awaiting triage bug java P2
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### What happened?

I am writing a pipeline to consume message from pubsub, do some validation, transform and sink to bigquery.

I need to load some data from external api call to be used in pipeline validation stage, for which I followed [Slowly updating global window side inputs](https://beam.apache.org/documentation/patterns/side-inputs/) to load config into side input. However, I keep getting error, while I already applied `Latest.globally()`:
```
Caused by: java.lang.IllegalArgumentException: PCollection with more than one element accessed as a singleton view. Consider using Combine.globally().asSingleton() to combine the PCollection into a single value

```

Going through online resource doesn't really help. However, I found that this happen only when the impulse duration is short, e.g. < 5s: GenerateSequence.from(0).withRate(1, **Duration.standardSeconds(1L)**))

**Is this expected?**

This bothers me, because I am not sure if the short period the root cause. Or will the error shows up again if the pipeline traffic becomes large even I have a minute as impulse period in product environment.

Any one have the same issue, or suggestion?

Beam version: 2.46.0

Here are simplied version my code, that can reproduce the error:
```java

import static com.applovin.array.silk.pipeline.common.Constants.CONFIG_DEV;
import static com.applovin.array.silk.pipeline.common.Constants.CONFIG_PROD;
import static org.apache.beam.sdk.options.SdkHarnessOptions.LogLevel;

import com.applovin.array.silk.pipeline.EventTransformer.EventTransform;
import com.applovin.array.silk.pipeline.EventTransformer.ExtractMessage;
import com.applovin.array.silk.pipeline.coders.FailureCoder;
import com.applovin.array.silk.pipeline.coders.JsonObjectCoder;
import com.applovin.array.silk.pipeline.coders.MessageContainerCoder;
import com.applovin.array.silk.pipeline.common.UncaughtExceptionLogger;
import com.applovin.array.silk.pipeline.models.Failure;
import com.applovin.array.silk.pipeline.models.MessageContainer;
import com.applovin.array.silk.pipeline.models.StartOptions;
import com.applovin.array.silk.pipeline.models.SystemConfigs;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonObject;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Latest;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.View;
import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
import org.apache.beam.sdk.transforms.windowing.Repeatedly;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TestSideInputPipeline {

private static final Logger logger = LoggerFactory.getLogger(TestSideInputPipeline.class);
private static final FailureCoder FAILURE_CODER = FailureCoder.of(
MessageContainerCoder.of());
private static final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
public static SystemConfigs systemConfigs;
private static final CloseableHttpClient httpClient = HttpClients.createDefault();

public static void main(String[] args) throws IOException {
UncaughtExceptionLogger.register();

StartOptions options = loadOptions(args);

run(options);
}

public static StartOptions loadOptions(String[] args) throws IOException {
StartOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(
StartOptions.class);

ClassLoader classLoader = TestSideInputPipeline.class.getClassLoader();
InputStream configStream = classLoader.getResourceAsStream(
"prod".equalsIgnoreCase(options.getEnv()) ? CONFIG_PROD : CONFIG_DEV);
systemConfigs = mapper.readValue(configStream, SystemConfigs.class);

options.setDefaultSdkHarnessLogLevel(LogLevel.valueOf(
systemConfigs.getLogLevel().toUpperCase()
));

return options;
}

public static void run(StartOptions options) {
Pipeline pipeline = Pipeline.create(options);
CoderRegistry coderRegistry = pipeline.getCoderRegistry();
coderRegistry.registerCoderForClass(JsonObject.class, JsonObjectCoder.of());
coderRegistry.registerCoderForType(FAILURE_CODER.getEncodedTypeDescriptor(), FAILURE_CODER);
coderRegistry.registerCoderForClass(MessageContainer.class, MessageContainerCoder.of());

// Pull Pubsub message with PubsubIO, which automatically acknowledge message after complete processing.
// https://cloud.google.com/dataflow/docs/concepts/streaming-with-cloud-pubsub
PCollection messages = pipeline.apply("ReadPubSubMessages",
PubsubIO.readMessagesWithAttributes()
.fromSubscription(systemConfigs.getIngestSubscription()));

// // extract message from pubsub structure
// ExtractMessage extractMessage = new ExtractMessage();
// PCollectionTuple extractMessageResult = messages.apply("ExtractMessage",
// ParDo.of(extractMessage).withOutputTags(extractMessage.getOutputTag(),
// TupleTagList.of(extractMessage.getFailuresTag())));
//
// // do some transform
// EventTransform eventTransform = new EventTransform();
// PCollectionTuple transformResult = extractMessageResult.get(extractMessage.getOutputTag())
// .apply("EventTransform", ParDo.of(eventTransform)
// .withOutputTags(eventTransform.getOutputTag(),
// TupleTagList.of(eventTransform.getFailuresTag())));
//
// PCollection transformedEvents = transformResult.get(
// eventTransform.getOutputTag());

// get side input view (json schemas), and validate
PCollectionView> map = pipeline.apply("Impulse",
GenerateSequence.from(0).withRate(1, Duration.standardSeconds(1)))
.apply(ParDo.of(new DoFn>() {
@ProcessElement
public void process(OutputReceiver> o) {
try {
CloseableHttpResponse response = httpClient.execute(
new HttpGet(systemConfigs.getSchemaUrl()));
String body = EntityUtils.toString(response.getEntity());
Map schema = mapper.readValue(body, Map.class);
logger.info("Loader got schema: {}", schema);
o.output(schema);
} catch (Exception e) {
logger.error("Failed, {}", e.getMessage());
o.output(Map.of());
}
}
}))
.apply(Window.>into(new GlobalWindows())
.triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
.discardingFiredPanes())
.apply(Latest.globally())
.apply(View.asSingleton());

PCollection validateEventResult = messages.apply("ValidateEvent",
ParDo.of(new DoFn() {
@ProcessElement
public void processElement(ProcessContext ctx) {
PubsubMessage message = ctx.element();
Map schemas = ctx.sideInput(map);
logger.info("Side input schemas: {}", schemas);

// validate message and save result.
ctx.output(message);
}
}).withSideInputs(map));

//more logic

pipeline.run();
}
}

```

### Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

### Issue Components

- [ ] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner

Contributor guide

Open the contributing guide

Research direction

Start with the simplified Java pipeline and reproduce the failure around GenerateSequence, GlobalWindows, Latest.globally(), and View.asSingleton(). Compare the short impulse period with the longer period described, then inspect the relevant Java SDK implementation and tests; done means determining whether the singleton error is expected and documenting or correcting the behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.