beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Neville Li (JIRA)" <j...@apache.org>
Subject [jira] [Updated] (BEAM-3234) PubsubIO batch size should be configurable
Date Tue, 21 Nov 2017 23:41:00 GMT

     [ https://issues.apache.org/jira/browse/BEAM-3234?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]

Neville Li updated BEAM-3234:
-----------------------------
    Description: 
Looks like there's a payload size limit in Pubsub, and PubsubIO has a hard coded batch size
that may cause this limit to be exceeded in some cases.

https://github.com/apache/beam/blob/release-2.1.0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java#L885

{code}
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptor;

public class Test {
  public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.create();
    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(GenerateSequence.from(0).to(100))
        .apply(MapElements
            .into(TypeDescriptor.of(String.class))
            .via(x -> {
              StringBuilder b = new StringBuilder();
              for (int i = 0; i < 10000000; i++) {
                b.append("x");
              }
              return b.toString();
            }))
        .apply(PubsubIO
            .writeStrings()
            .to("projects/scio-playground/topics/payload-test"));

    pipeline.run().waitUntilFinish();
  }
}
{code}

The above code throws the following error:
{code}
[error] Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 400
Bad Request
[error] {
[error]   "code" : 400,
[error]   "errors" : [ {
[error]     "domain" : "global",
[error]     "message" : "Request payload size exceeds the limit: 10485760 bytes.",
[error]     "reason" : "badRequest"
[error]   } ],
[error]   "message" : "Request payload size exceeds the limit: 10485760 bytes.",
[error]   "status" : "INVALID_ARGUMENT"
{code}

  was:
Looks like there's a payload size limit in Pubsub, and PubsubIO has a hard coded batch size
that may cause this limit to be exceeded in some cases.

{{{
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptor;

public class Test {
  public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.create();
    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(GenerateSequence.from(0).to(100))
        .apply(MapElements
            .into(TypeDescriptor.of(String.class))
            .via(x -> {
              StringBuilder b = new StringBuilder();
              for (int i = 0; i < 10000000; i++) {
                b.append("x");
              }
              return b.toString();
            }))
        .apply(PubsubIO
            .writeStrings()
            .to("projects/scio-playground/topics/payload-test"));

    pipeline.run().waitUntilFinish();
  }
}
}}}

The above code throws the following error:
{{{
[error] Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 400
Bad Request
[error] {
[error]   "code" : 400,
[error]   "errors" : [ {
[error]     "domain" : "global",
[error]     "message" : "Request payload size exceeds the limit: 10485760 bytes.",
[error]     "reason" : "badRequest"
[error]   } ],
[error]   "message" : "Request payload size exceeds the limit: 10485760 bytes.",
[error]   "status" : "INVALID_ARGUMENT"
}}}


> PubsubIO batch size should be configurable
> ------------------------------------------
>
>                 Key: BEAM-3234
>                 URL: https://issues.apache.org/jira/browse/BEAM-3234
>             Project: Beam
>          Issue Type: Bug
>          Components: sdk-java-gcp
>    Affects Versions: 2.1.0
>            Reporter: Neville Li
>            Priority: Minor
>
> Looks like there's a payload size limit in Pubsub, and PubsubIO has a hard coded batch
size that may cause this limit to be exceeded in some cases.
> https://github.com/apache/beam/blob/release-2.1.0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java#L885
> {code}
> import org.apache.beam.sdk.Pipeline;
> import org.apache.beam.sdk.io.GenerateSequence;
> import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
> import org.apache.beam.sdk.options.PipelineOptions;
> import org.apache.beam.sdk.options.PipelineOptionsFactory;
> import org.apache.beam.sdk.transforms.MapElements;
> import org.apache.beam.sdk.values.TypeDescriptor;
> public class Test {
>   public static void main(String[] args) {
>     PipelineOptions options = PipelineOptionsFactory.create();
>     Pipeline pipeline = Pipeline.create(options);
>     pipeline
>         .apply(GenerateSequence.from(0).to(100))
>         .apply(MapElements
>             .into(TypeDescriptor.of(String.class))
>             .via(x -> {
>               StringBuilder b = new StringBuilder();
>               for (int i = 0; i < 10000000; i++) {
>                 b.append("x");
>               }
>               return b.toString();
>             }))
>         .apply(PubsubIO
>             .writeStrings()
>             .to("projects/scio-playground/topics/payload-test"));
>     pipeline.run().waitUntilFinish();
>   }
> }
> {code}
> The above code throws the following error:
> {code}
> [error] Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException:
400 Bad Request
> [error] {
> [error]   "code" : 400,
> [error]   "errors" : [ {
> [error]     "domain" : "global",
> [error]     "message" : "Request payload size exceeds the limit: 10485760 bytes.",
> [error]     "reason" : "badRequest"
> [error]   } ],
> [error]   "message" : "Request payload size exceeds the limit: 10485760 bytes.",
> [error]   "status" : "INVALID_ARGUMENT"
> {code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message