apache_beam.io.requestresponse module

PTransform for reading from and writing to Web APIs.

class apache_beam.io.requestresponse.ExponentialBackOffRepeater[source]

Bases: apache_beam.io.requestresponse.Repeater

Configure exponential backoff retry strategy.

It retries for exceptions due to the remote service such as TooManyRequests (HTTP 429), UserCodeTimeoutException, UserCodeQuotaException.

It utilizes the decorator apache_beam.utils.retry.with_exponential_backoff().

repeat(caller: apache_beam.io.requestresponse.Caller[~RequestT, ~ResponseT][RequestT, ResponseT], request: RequestT, timeout: float, metrics_collector: Optional[apache_beam.io.requestresponse._MetricsCollector] = None) → ResponseT[source]

repeat method is called from the RequestResponseIO when a repeater is enabled.

  • caller – a ~apache_beam.io.requestresponse.Caller object that calls the API.
  • request – input request to repeat.
  • timeout – time to wait for the request to complete.
  • metrics_collector – (Optional) a ~apache_beam.io.requestresponse._MetricsCollector object to collect the metrics for RequestResponseIO.
class apache_beam.io.requestresponse.NoOpsRepeater[source]

Bases: apache_beam.io.requestresponse.Repeater

Executes a request just once irrespective of any exception.

repeat(caller: apache_beam.io.requestresponse.Caller[~RequestT, ~ResponseT][RequestT, ResponseT], request: RequestT, timeout: float, metrics_collector: Optional[apache_beam.io.requestresponse._MetricsCollector]) → ResponseT[source]
class apache_beam.io.requestresponse.DefaultThrottler(window_ms: int = 1, bucket_ms: int = 1, overload_ratio: float = 2, delay_secs: int = 5)[source]

Bases: apache_beam.io.requestresponse.PreCallThrottler

Default throttler that uses apache_beam.io.components.adaptive_throttler.AdaptiveThrottler

  • window_ms (int) – length of history to consider, in ms, to set throttling.
  • bucket_ms (int) – granularity of time buckets that we store data in, in ms.
  • overload_ratio (float) – the target ratio between requests sent and successful requests. This is “K” in the formula in https://landing.google.com/sre/book/chapters/handling-overload.html.
  • delay_secs (int) – minimum number of seconds to throttle a request.
class apache_beam.io.requestresponse.RedisCache(host: str, port: int, time_to_live: Union[int, datetime.timedelta] = 86400, *, request_coder: Optional[apache_beam.coders.coders.Coder] = None, response_coder: Optional[apache_beam.coders.coders.Coder] = None, **kwargs)[source]

Bases: apache_beam.io.requestresponse.Cache

Configure cache using Redis for apache_beam.io.requestresponse.RequestResponseIO.

  • host (str) – The hostname or IP address of the Redis server.
  • port (int) – The port number of the Redis server.
  • time_to_live(Union[int, timedelta]) The time-to-live (TTL) for records stored in Redis. Provide an integer (in seconds) or a datetime.timedelta object.
  • request_coder – (Optional[coders.Coder]) coder for encoding requests.
  • response_coder – (Optional[coders.Coder]) coder for decoding responses received from Redis.
  • kwargs – Optional additional keyword arguments that are required to connect to your redis server. Same as redis.Redis().

get_read returns a PTransform for reading from the cache.


returns a PTransform for writing to the cache.

class apache_beam.io.requestresponse.RequestResponseIO(caller: apache_beam.io.requestresponse.Caller[~RequestT, ~ResponseT][RequestT, ResponseT], timeout: Optional[float] = 30, should_backoff: Optional[apache_beam.io.requestresponse.ShouldBackOff] = None, repeater: apache_beam.io.requestresponse.Repeater = <apache_beam.io.requestresponse.ExponentialBackOffRepeater object>, cache: Optional[apache_beam.io.requestresponse.Cache] = None, throttler: apache_beam.io.requestresponse.PreCallThrottler = <apache_beam.io.requestresponse.DefaultThrottler object>)[source]

Bases: apache_beam.transforms.ptransform.PTransform

A RequestResponseIO transform to read and write to APIs.

Processes an input PCollection of requests by making a call to the API as defined in Caller’s __call__ method and returns a PCollection of responses.

Instantiates a RequestResponseIO transform.

expand(requests: apache_beam.pvalue.PCollection[~RequestT][RequestT]) → apache_beam.pvalue.PCollection[~ResponseT][ResponseT][source]