Apache Beam Split to Multiple Pipeline Output

Question

I am subscribing from one topic and contains different event types and they pass in with different attributes.

After I read the element, based on their attribute, I need to move them to different places. This is the sample code look like:

    Options options =PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    Pipeline pipeline = Pipeline.create(options);
    pipeline
    .apply(
        "ReadType1",
        EventIO.<T>readJsons()
            .of(T.class)
            .withPubsubTimestampAttributeName(null)
            .withOptions(options))
    .apply(
        Filter.by(
            new SerializableFunction<T, Boolean>() {
              @Override
              public Boolean apply(T input) {
                return input.attributes.get("type").equals("type1");
              }
            }))
    .apply(
        "WindowMetrics",
        Window.into(FixedWindows.of(Duration.standardSeconds(options.getWindowDuration()))))
    .apply("AsJsons", AsJsons.of(T.class))
    .apply(
        "Write File(s)",
        TextIO.write()
            .withWindowedWrites()
            .withNumShards(options.getNumShards())
            .to(
                new WindowedFilenamePolicy(
                    options.getRunOutputDirectory(),
                    options.getUseCurrentDateForOutputDirectory(),
                    options.getOutputFilenamePrefix(),
                    options.getOutputShardTemplate(),
                    options.getOutputFilenameSuffix()))
            .withTempDirectory(
                NestedValueProvider.of(
                    options.getTempDirectory(),
                    (SerializableFunction<String, ResourceId>)
                        input -> FileBasedSink.convertToFileResourceIfPossible(input))));


pipeline.apply("ReadType2",
        EventIO.<T>readJsons().of(T.class)
                .withPubsubTimestampAttributeName(null)
                .withOptions(options))
        .apply(Filter.by(new SerializableFunction<T, Boolean>() {
          @Override
          public Boolean apply(Event input) {
            return input.attributes.get("type").equals("type2");
          }
        })).apply( "WindowMetrics",
        Window.into(FixedWindows.of(Duration.standardSeconds(options.getWindowDuration()))))
        .apply("AsJsons", AsJsons.of(T.class))
        .apply(
                "Write File(s)",
                TextIO.write()
                        .withWindowedWrites()
                        .withNumShards(options.getNumShards())
                        .to(
                                new WindowedFilenamePolicy(
                                        options.getBatchOutputDirectory(),
                                        options.getUseCurrentDateForOutputDirectory(),
                                        options.getOutputFilenamePrefix(),
                                        options.getOutputShardTemplate(),
                                        options.getOutputFilenameSuffix()))
                        .withTempDirectory(
                                NestedValueProvider.of(
                                        options.getTempDirectory(),
                                        (SerializableFunction<String, ResourceId>)
                                                input -> FileBasedSink.convertToFileResourceIfPossible(input))));

pipeline.apply("ReadType3",
        EventIO.<Event>readJsons().of(T.class)
                .withPubsubTimestampAttributeName(null)
                .withOptions(options))
        .apply(Filter.by(new SerializableFunction<T, Boolean>() {
          @Override
          public Boolean apply(T input) {
            return input.attributes.get("type").equals("type3");
          }
        })).apply( "WindowMetrics",
        Window.into(FixedWindows.of(Duration.standardSeconds(options.getWindowDuration()))))
        .apply("AsJsons", AsJsons.of(T.class))
        .apply(
                "Write File(s)",
                TextIO.write()
                        .withWindowedWrites()
                        .withNumShards(options.getNumShards())
                        .to(
                                new WindowedFilenamePolicy(
                                        options.getCustomIntervalOutputDirectory(),
                                        options.getUseCurrentDateForOutputDirectory(),
                                        options.getOutputFilenamePrefix(),
                                        options.getOutputShardTemplate(),
                                        options.getOutputFilenameSuffix()))
                        .withTempDirectory(
                                NestedValueProvider.of(
                                        options.getTempDirectory(),
                                        (SerializableFunction<String, ResourceId>)
                                                input -> FileBasedSink.convertToFileResourceIfPossible(input))));

pipeline.run();

Basically I read an event and filter them on their attribute and write the file. The job failed in dataflow as Workflow failed. Causes: The pubsub configuration contains errors: Subscription 'sub-name' is consumed by multiple stages, this will result in undefined behavior.

So what will be the appropriate way to split the pipeline within the same job?

I tried Pipeline1, Pipeline2,Pipeline3 and it end up need to multiple job name to run multiple pipeline, I am not sure that should the right way to do it.

Brent Worden · Accepted Answer · 2022-04-13 01:39:34Z

The two EventIO transforms on the same subscription are the cause of the error. You need to eliminate one of those transforms in order for this to work. This can be done by consuming the subscription into a single PCollection and then applying two filtering branches to that collection individually. Here is a partial example:

// single PCollection of the events consumed from the subscription
PCollection<T> events = pipeline
  .apply("Read Events",
    EventIO.<T>readJsons()
      .of(T.class)
      .withPubsubTimestampAttributeName(null)
      .withOptions(options));

// PCollection of type1 events
PCollection<T> typeOneEvents = events.apply(
  Filter.by(
    new SerializableFunction<T, Boolean>() {
      @Override
      public Boolean apply(T input) {
        return input.attributes.get("type").equals("type1");
      }}));
// TODO typeOneEvents.apply("WindowMetrics / AsJsons / Write File(s)")

// PCollection of type2 events
PCollection<T> typeTwoEvents = events.apply(
  Filter.by(
    new SerializableFunction<T, Boolean>() {
      @Override
      public Boolean apply(T input) {
        return input.attributes.get("type").equals("type2");
      }}));
// TODO typeTwoEvents.apply("WindowMetrics / AsJsons / Write File(s)")

Another possibility is to use some other transforms provided by Apache Beam. Doing so might simplify your solution a little. Once such transform is Partition. Partition allows for the splitting of a single PCollection in a fixed number of PCollections based on a partitioning function. A partial example using Partition is:

// single PCollection of the events consumed from the subscription
PCollectionList<T> eventsByType = pipeline
  .apply("Read Events",
    EventIO.<T>readJsons()
      .of(T.class)
      .withPubsubTimestampAttributeName(null)
      .withOptions(options))
  .apply("Partition By Type",
    Partition.of(2, new PartitionFn<T>() {
      public int partitionFor(T event, int numPartitions) {
        return input.attributes.get("type").equals("type1") ? 0 : 1;
      }}));

PCollection<T> typeOneEvents = eventsByType.get(0);
// TODO typeOneEvents.apply("WindowMetrics / AsJsons / Write File(s)")

PCollection<T> typeTwoEvents = eventsByType.get(1);
// TODO typeTwoEvents.apply("WindowMetrics / AsJsons / Write File(s)")

Jie Zhang · Accepted Answer · 2022-04-12 19:01:05Z

1

The answer should be using Partition in Beam.

https://beam.apache.org/documentation/transforms/java/elementwise/partition/

answered Apr 12, 2022 at 19:01

Jie Zhang

133 bronze badges

1 Comment

Community Over a year ago

Your answer could be improved with additional supporting information. Please edit to add further details, such as citations or documentation, so that others can confirm that your answer is correct. You can find more information on how to write good answers in the help center.

Collectives™ on Stack Overflow

Apache Beam Split to Multiple Pipeline Output

2 Answers 2

Comments

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related