apache_beam.io.iobase module¶
Sources and sinks.
A Source manages record-oriented data input from a particular kind of source (e.g. a set of files, a database table, etc.). The reader() method of a source returns a reader object supporting the iterator protocol; iteration yields raw records of unprocessed, serialized data.
A Sink manages record-oriented data output to a particular kind of sink (e.g. a set of files, a database table, etc.). The writer() method of a sink returns a writer object supporting writing records of serialized data to the sink.
-
class
apache_beam.io.iobase.
BoundedSource
[source]¶ Bases:
apache_beam.io.iobase.SourceBase
A source that reads a finite amount of input records.
This class defines following operations which can be used to read the source efficiently.
- Size estimation - method
estimate_size()
may return an accurate estimation in bytes for the size of the source. - Splitting into bundles of a given size - method
split()
can be used to split the source into a set of sub-sources (bundles) based on a desired bundle size. - Getting a RangeTracker - method
get_range_tracker()
should return aRangeTracker
object for a given position range for the position type of the records returned by the source. - Reading the data - method
read()
can be used to read data from the source while respecting the boundaries defined by a givenRangeTracker
.
A runner will perform reading the source in two steps.
- Method
get_range_tracker()
will be invoked with start and end positions to obtain aRangeTracker
for the range of positions the runner intends to read. Source must define a default initial start and end position range. These positions must be used if the start and/or end positions passed to the methodget_range_tracker()
areNone
- Method read() will be invoked with the
RangeTracker
obtained in the previous step.
Mutability
A
BoundedSource
object should not be mutated while its methods (for example,read()
) are being invoked by a runner. Runner implementations may invoke methods ofBoundedSource
objects through multi-threaded and/or reentrant execution modes.-
estimate_size
() → Optional[int][source]¶ Estimates the size of source in bytes.
An estimate of the total size (in bytes) of the data that would be read from this source. This estimate is in terms of external storage size, before performing decompression or other processing.
Returns: estimated size of the source if the size can be determined, None
otherwise.
-
split
(desired_bundle_size: int, start_position: Optional[Any] = None, stop_position: Optional[Any] = None) → Iterator[apache_beam.io.iobase.SourceBundle][source]¶ Splits the source into a set of bundles.
Bundles should be approximately of size
desired_bundle_size
bytes.Parameters: - desired_bundle_size – the desired size (in bytes) of the bundles returned.
- start_position – if specified the given position must be used as the starting position of the first bundle.
- stop_position – if specified the given position must be used as the ending position of the last bundle.
Returns: an iterator of objects of type ‘SourceBundle’ that gives information about the generated bundles.
-
get_range_tracker
(start_position: Optional[Any], stop_position: Optional[Any]) → apache_beam.io.iobase.RangeTracker[source]¶ Returns a RangeTracker for a given position range.
Framework may invoke
read()
method with the RangeTracker object returned here to read data from the source.Parameters: - start_position – starting position of the range. If ‘None’ default start position of the source must be used.
- stop_position – ending position of the range. If ‘None’ default stop position of the source must be used.
Returns: a
RangeTracker
for the given position range.
-
read
(range_tracker)[source]¶ Returns an iterator that reads data from the source.
The returned set of data must respect the boundaries defined by the given
RangeTracker
object. For example:- Returned set of data must be for the range
[range_tracker.start_position, range_tracker.stop_position)
. Note that a source may decide to return records that start afterrange_tracker.stop_position
. See documentation in classRangeTracker
for more details. Also, note that framework might invokerange_tracker.try_split()
to perform dynamic split operations. range_tracker.stop_position may be updated dynamically due to successful dynamic split operations. - Method
range_tracker.try_split()
must be invoked for every record that starts at a split point. - Method
range_tracker.record_current_position()
may be invoked for records that do not start at split points.
Parameters: range_tracker – a RangeTracker
whose boundaries must be respected when reading data from the source. A runner that reads this source muss pass aRangeTracker
object that is notNone
.Returns: an iterator of data read by the source. - Returned set of data must be for the range
- Size estimation - method
-
class
apache_beam.io.iobase.
RangeTracker
[source]¶ Bases:
object
A thread safe object used by Dataflow source framework.
A Dataflow source is defined using a ‘’BoundedSource’’ and a ‘’RangeTracker’’ pair. A ‘’RangeTracker’’ is used by Dataflow source framework to perform dynamic work rebalancing of position-based sources.
Position-based sources
A position-based source is one where the source can be described by a range of positions of an ordered type and the records returned by the reader can be described by positions of the same type.
In case a record occupies a range of positions in the source, the most important thing about the record is the position where it starts.
Defining the semantics of positions for a source is entirely up to the source class, however the chosen definitions have to obey certain properties in order to make it possible to correctly split the source into parts, including dynamic splitting. Two main aspects need to be defined:
- How to assign starting positions to records.
- Which records should be read by a source with a range ‘[A, B)’.
Moreover, reading a range must be efficient, i.e., the performance of reading a range should not significantly depend on the location of the range. For example, reading the range [A, B) should not require reading all data before ‘A’.
The sections below explain exactly what properties these definitions must satisfy, and how to use a
RangeTracker
with a properly defined source.Properties of position-based sources
The main requirement for position-based sources is associativity: reading records from ‘[A, B)’ and records from ‘[B, C)’ should give the same records as reading from ‘[A, C)’, where ‘A <= B <= C’. This property ensures that no matter how a range of positions is split into arbitrarily many sub-ranges, the total set of records described by them stays the same.
The other important property is how the source’s range relates to positions of records in the source. In many sources each record can be identified by a unique starting position. In this case:
- All records returned by a source ‘[A, B)’ must have starting positions in this range.
- All but the last record should end within this range. The last record may or may not extend past the end of the range.
- Records should not overlap.
Such sources should define “read ‘[A, B)’” as “read from the first record starting at or after ‘A’, up to but not including the first record starting at or after ‘B’”.
Some examples of such sources include reading lines or CSV from a text file, reading keys and values from a BigTable, etc.
The concept of split points allows to extend the definitions for dealing with sources where some records cannot be identified by a unique starting position.
In all cases, all records returned by a source ‘[A, B)’ must start at or after ‘A’.
Split points
Some sources may have records that are not directly addressable. For example, imagine a file format consisting of a sequence of compressed blocks. Each block can be assigned an offset, but records within the block cannot be directly addressed without decompressing the block. Let us refer to this hypothetical format as <i>CBF (Compressed Blocks Format)</i>.
Many such formats can still satisfy the associativity property. For example, in CBF, reading ‘[A, B)’ can mean “read all the records in all blocks whose starting offset is in ‘[A, B)’”.
To support such complex formats, we introduce the notion of split points. We say that a record is a split point if there exists a position ‘A’ such that the record is the first one to be returned when reading the range ‘[A, infinity)’. In CBF, the only split points would be the first records in each block.
Split points allow us to define the meaning of a record’s position and a source’s range in all cases:
- For a record that is at a split point, its position is defined to be the largest ‘A’ such that reading a source with the range ‘[A, infinity)’ returns this record.
- Positions of other records are only required to be non-decreasing.
- Reading the source ‘[A, B)’ must return records starting from the first split point at or after ‘A’, up to but not including the first split point at or after ‘B’. In particular, this means that the first record returned by a source MUST always be a split point.
- Positions of split points must be unique.
As a result, for any decomposition of the full range of the source into position ranges, the total set of records will be the full set of records in the source, and each record will be read exactly once.
Consumed positions
As the source is being read, and records read from it are being passed to the downstream transforms in the pipeline, we say that positions in the source are being consumed. When a reader has read a record (or promised to a caller that a record will be returned), positions up to and including the record’s start position are considered consumed.
Dynamic splitting can happen only at unconsumed positions. If the reader just returned a record at offset 42 in a file, dynamic splitting can happen only at offset 43 or beyond, as otherwise that record could be read twice (by the current reader and by a reader of the task starting at 43).
-
SPLIT_POINTS_UNKNOWN
= <object object>¶
-
try_claim
(position)[source]¶ Atomically determines if a record at a split point is within the range.
This method should be called if and only if the record is at a split point. This method may modify the internal state of the
RangeTracker
by updating the last-consumed position toposition
.** Thread safety **
Methods of the class
RangeTracker
including this method may get invoked by different threads, hence must be made thread-safe, e.g. by using a single lock object.Parameters: position – starting position of a record being read by a source. Returns: True
, if the given position falls within the current range, returnsFalse
otherwise.
-
set_current_position
(position)[source]¶ Updates the last-consumed position to the given position.
A source may invoke this method for records that do not start at split points. This may modify the internal state of the
RangeTracker
. If the record starts at a split point, methodtry_claim()
must be invoked instead of this method.Parameters: position – starting position of a record being read by a source.
-
position_at_fraction
(fraction)[source]¶ Returns the position at the given fraction.
Given a fraction within the range [0.0, 1.0) this method will return the position at the given fraction compared to the position range [self.start_position, self.stop_position).
** Thread safety **
Methods of the class
RangeTracker
including this method may get invoked by different threads, hence must be made thread-safe, e.g. by using a single lock object.Parameters: fraction – a float value within the range [0.0, 1.0). Returns: a position within the range [self.start_position, self.stop_position).
-
try_split
(position)[source]¶ Atomically splits the current range.
Determines a position to split the current range, split_position, based on the given position. In most cases split_position and position will be the same.
Splits the current range ‘[self.start_position, self.stop_position)’ into a “primary” part ‘[self.start_position, split_position)’ and a “residual” part ‘[split_position, self.stop_position)’, assuming the current last-consumed position is within ‘[self.start_position, split_position)’ (i.e., split_position has not been consumed yet).
If successful, updates the current range to be the primary and returns a tuple (split_position, split_fraction). split_fraction should be the fraction of size of range ‘[self.start_position, split_position)’ compared to the original (before split) range ‘[self.start_position, self.stop_position)’.
If the split_position has already been consumed, returns
None
.** Thread safety **
Methods of the class
RangeTracker
including this method may get invoked by different threads, hence must be made thread-safe, e.g. by using a single lock object.Parameters: position – suggested position where the current range should try to be split at. Returns: a tuple containing the split position and split fraction if split is successful. Returns None
otherwise.
-
fraction_consumed
()[source]¶ Returns the approximate fraction of consumed positions in the source.
** Thread safety **
Methods of the class
RangeTracker
including this method may get invoked by different threads, hence must be made thread-safe, e.g. by using a single lock object.Returns: the approximate fraction of positions that have been consumed by successful ‘try_split()’ and ‘try_claim()’ calls, or 0.0 if no such calls have happened.
-
split_points
()[source]¶ Gives the number of split points consumed and remaining.
For a
RangeTracker
used by aBoundedSource
(within aBoundedSource.read()
invocation) this method produces a 2-tuple that gives the number of split points consumed by theBoundedSource
and the number of split points remaining within the range of theRangeTracker
that has not been consumed by theBoundedSource
.More specifically, given that the position of the current record being read by
BoundedSource
is current_position this method produces a tuple that consists of (1) number of split points in the range [self.start_position(), current_position) without including the split point that is currently being consumed. This represents the total amount of parallelism in the consumed part of the source. (2) number of split points within the range [current_position, self.stop_position()) including the split point that is currently being consumed. This represents the total amount of parallelism in the unconsumed part of the source.Methods of the class
RangeTracker
including this method may get invoked by different threads, hence must be made thread-safe, e.g. by using a single lock object.- ** General information about consumed and remaining number of split
- points returned by this method. **
- Before a source read (
BoundedSource.read()
invocation) claims the first split point, number of consumed split points is 0. This condition holds independent of whether the input is “splittable”. A splittable source is a source that has more than one split point. - Any source read that has only claimed one split point has 0 consumed split points since the first split point is the current split point and is still being processed. This condition holds independent of whether the input is splittable.
- For an empty source read which never invokes
RangeTracker.try_claim()
, the consumed number of split points is 0. This condition holds independent of whether the input is splittable. - For a source read which has invoked
RangeTracker.try_claim()
n times, the consumed number of split points is n -1. - If a
BoundedSource
sets a callback through functionset_split_points_unclaimed_callback()
,RangeTracker
can use that callback when determining remaining number of split points. - Remaining split points should include the split point that is currently being consumed by the source read. Hence if the above callback returns an integer value n, remaining number of split points should be (n + 1).
- After last split point is claimed remaining split points becomes 1, because this unfinished read itself represents an unfinished split point.
- After all records of the source has been consumed, remaining number of split points becomes 0 and consumed number of split points becomes equal to the total number of split points within the range being read by the source. This method does not address this condition and will continue to report number of consumed split points as (“total number of split points” - 1) and number of remaining split points as 1. A runner that performs the reading of the source can detect when all records have been consumed and adjust remaining and consumed number of split points accordingly.
- Before a source read (
** Examples **
A “perfectly splittable” input which can be read in parallel down to the individual records.
Consider a perfectly splittable input that consists of 50 split points.
- Before a source read (
BoundedSource.read()
invocation) claims the first split point, number of consumed split points is 0 number of remaining split points is 50. - After claiming first split point, consumed number of split points is 0 and remaining number of split is 50.
- After claiming split point #30, consumed number of split points is 29 and remaining number of split points is 21.
- After claiming all 50 split points, consumed number of split points is 49 and remaining number of split points is 1.
a “block-compressed” file format such as
avroio
, in which a block of records has to be read as a whole, but different blocks can be read in parallel.Consider a block compressed input that consists of 5 blocks.
- Before a source read (
BoundedSource.read()
invocation) claims the first split point (first block), number of consumed split points is 0 number of remaining split points is 5. - After claiming first split point, consumed number of split points is 0 and remaining number of split is 5.
- After claiming split point #3, consumed number of split points is 2 and remaining number of split points is 3.
- After claiming all 5 split points, consumed number of split points is 4 and remaining number of split points is 1.
an “unsplittable” input such as a cursor in a database or a gzip compressed file.
Such an input is considered to have only a single split point. Number of consumed split points is always 0 and number of remaining split points is always 1.
By default
RangeTracker` returns ``RangeTracker.SPLIT_POINTS_UNKNOWN
for both consumed and remaining number of split points, which indicates that the number of split points consumed and remaining is unknown.Returns: A pair that gives consumed and remaining number of split points. Consumed number of split points should be an integer larger than or equal to zero or RangeTracker.SPLIT_POINTS_UNKNOWN
. Remaining number of split points should be an integer larger than zero orRangeTracker.SPLIT_POINTS_UNKNOWN
.
-
set_split_points_unclaimed_callback
(callback)[source]¶ Sets a callback for determining the unclaimed number of split points.
By invoking this function, a
BoundedSource
can set a callback function that may get invoked by theRangeTracker
to determine the number of unclaimed split points. A split point is unclaimed ifRangeTracker.try_claim()
method has not been successfully invoked for that particular split point. The callback function accepts a single parameter, a stop position for the BoundedSource (stop_position). If the record currently being consumed by theBoundedSource
is at position current_position, callback should return the number of split points within the range (current_position, stop_position). Note that, this should not include the split point that is currently being consumed by the source.This function must be implemented by subclasses before being used.
Parameters: callback – a function that takes a single parameter, a stop position, and returns unclaimed number of split points for the source read operation that is calling this function. Value returned from callback should be either an integer larger than or equal to zero or RangeTracker.SPLIT_POINTS_UNKNOWN
.
-
class
apache_beam.io.iobase.
Sink
[source]¶ Bases:
apache_beam.transforms.display.HasDisplayData
This class is deprecated, no backwards-compatibility guarantees.
A resource that can be written to using the
beam.io.Write
transform.Here
beam
stands for Apache Beam Python code imported in following manner.import apache_beam as beam
.A parallel write to an
iobase.Sink
consists of three phases:- A sequential initialization phase (e.g., creating a temporary output directory, etc.)
- A parallel write phase where workers write bundles of records
- A sequential finalization phase (e.g., committing the writes, merging output files, etc.)
Implementing a new sink requires extending two classes.
- iobase.Sink
iobase.Sink
is an immutable logical description of the location/resource to write to. Depending on the type of sink, it may contain fields such as the path to an output directory on a filesystem, a database table name, etc.iobase.Sink
provides methods for performing a write operation to the sink described by it. To this end, implementors of an extension ofiobase.Sink
must implement three methods:initialize_write()
,open_writer()
, andfinalize_write()
.- iobase.Writer
iobase.Writer
is used to write a single bundle of records. Aniobase.Writer
defines two methods:write()
which writes a single record from the bundle andclose()
which is called once at the end of writing a bundle.See also
apache_beam.io.filebasedsink.FileBasedSink
which provides a simpler API for writing sinks that produce files.Execution of the Write transform
initialize_write()
,pre_finalize()
, andfinalize_write()
are conceptually called once. However, implementors must ensure that these methods are idempotent, as they may be called multiple times on different machines in the case of failure/retry. A method may be called more than once concurrently, in which case it’s okay to have a transient failure (such as due to a race condition). This failure should not prevent subsequent retries from succeeding.initialize_write()
should perform any initialization that needs to be done prior to writing to the sink.initialize_write()
may return a result (let’s call thisinit_result
) that contains any parameters it wants to pass on to its writers about the sink. For example, a sink that writes to a file system may return aninit_result
that contains a dynamically generated unique directory to which data should be written.To perform writing of a bundle of elements, Dataflow execution engine will create an
iobase.Writer
using the implementation ofiobase.Sink.open_writer()
. When invokingopen_writer()
execution engine will provide theinit_result
returned byinitialize_write()
invocation as well as a bundle id (let’s call thisbundle_id
) that is unique for each invocation ofopen_writer()
.Execution engine will then invoke
iobase.Writer.write()
implementation for each element that has to be written. Once all elements of a bundle are written, execution engine will invokeiobase.Writer.close()
implementation which should return a result (let’s call thiswrite_result
) that contains information that encodes the result of the write and, in most cases, some encoding of the unique bundle id. For example, if each bundle is written to a unique temporary file,close()
method may return an object that contains the temporary file name. After writing of all bundles is complete, execution engine will invokepre_finalize()
and thenfinalize_write()
implementation.The execution of a write transform can be illustrated using following pseudo code (assume that the outer for loop happens in parallel across many machines):
init_result = sink.initialize_write() write_results = [] for bundle in partition(pcoll): writer = sink.open_writer(init_result, generate_bundle_id()) for elem in bundle: writer.write(elem) write_results.append(writer.close()) pre_finalize_result = sink.pre_finalize(init_result, write_results) sink.finalize_write(init_result, write_results, pre_finalize_result)
init_result
Methods of ‘iobase.Sink’ should agree on the ‘init_result’ type that will be returned when initializing the sink. This type can be a client-defined object or an existing type. The returned type must be picklable using Dataflow coder
coders.PickleCoder
. Returning an init_result is optional.bundle_id
In order to ensure fault-tolerance, a bundle may be executed multiple times (e.g., in the event of failure/retry or for redundancy). However, exactly one of these executions will have its result passed to the
iobase.Sink.finalize_write()
method. Each call toiobase.Sink.open_writer()
is passed a unique bundle id when it is called by theWriteImpl
transform, so even redundant or retried bundles will have a unique way of identifying their output.The bundle id should be used to guarantee that a bundle’s output is unique. This uniqueness guarantee is important; if a bundle is to be output to a file, for example, the name of the file must be unique to avoid conflicts with other writers. The bundle id should be encoded in the writer result returned by the writer and subsequently used by the
finalize_write()
method to identify the results of successful writes.For example, consider the scenario where a Writer writes files containing serialized records and the
finalize_write()
is to merge or rename these output files. In this case, a writer may use its unique id to name its output file (to avoid conflicts) and return the name of the file it wrote as its writer result. Thefinalize_write()
will then receive anIterable
of output file names that it can then merge or rename using some bundle naming scheme.write_result
iobase.Writer.close()
andfinalize_write()
implementations must agree on type of thewrite_result
object returned when invokingiobase.Writer.close()
. This type can be a client-defined object or an existing type. The returned type must be picklable using Dataflow codercoders.PickleCoder
. Returning awrite_result
wheniobase.Writer.close()
is invoked is optional but if uniquewrite_result
objects are not returned, sink should, guarantee idempotency when same bundle is written multiple times due to failure/retry or redundancy.More information
For more information on creating new sinks please refer to the official documentation at
https://beam.apache.org/documentation/sdks/python-custom-io#creating-sinks
-
skip_if_empty
= False¶
-
initialize_write
()[source]¶ Initializes the sink before writing begins.
Invoked before any data is written to the sink.
Please see documentation in
iobase.Sink
for an example.Returns: An object that contains any sink specific state generated by initialization. This object will be passed to open_writer() and finalize_write() methods.
-
open_writer
(init_result, uid)[source]¶ Opens a writer for writing a bundle of elements to the sink.
Parameters: - init_result – the result of initialize_write() invocation.
- uid – a unique identifier generated by the system.
Returns: an
iobase.Writer
that can be used to write a bundle of records to the current sink.
-
pre_finalize
(init_result, writer_results)[source]¶ Pre-finalization stage for sink.
Called after all bundle writes are complete and before finalize_write. Used to setup and verify filesystem and sink states.
Parameters: - init_result – the result of
initialize_write()
invocation. - writer_results – an iterable containing results of
Writer.close()
invocations. This will only contain results of successful writes, and will only contain the result of a single successful write for a given bundle.
Returns: An object that contains any sink specific state generated. This object will be passed to finalize_write().
- init_result – the result of
-
finalize_write
(init_result, writer_results, pre_finalize_result)[source]¶ Finalizes the sink after all data is written to it.
Given the result of initialization and an iterable of results from bundle writes, performs finalization after writing and closes the sink. Called after all bundle writes are complete.
The bundle write results that are passed to finalize are those returned by bundles that completed successfully. Although bundles may have been run multiple times (for fault-tolerance), only one writer result will be passed to finalize for each bundle. An implementation of finalize should perform clean up of any failed and successfully retried bundles. Note that these failed bundles will not have their writer result passed to finalize, so finalize should be capable of locating any temporary/partial output written by failed bundles.
If all retries of a bundle fails, the whole pipeline will fail without finalize_write() being invoked.
A best practice is to make finalize atomic. If this is impossible given the semantics of the sink, finalize should be idempotent, as it may be called multiple times in the case of failure/retry or for redundancy.
Note that the iteration order of the writer results is not guaranteed to be consistent if finalize is called multiple times.
Parameters: - init_result – the result of
initialize_write()
invocation. - writer_results – an iterable containing results of
Writer.close()
invocations. This will only contain results of successful writes, and will only contain the result of a single successful write for a given bundle. - pre_finalize_result – the result of
pre_finalize()
invocation.
- init_result – the result of
-
class
apache_beam.io.iobase.
Writer
[source]¶ Bases:
object
This class is deprecated, no backwards-compatibility guarantees.
Writes a bundle of elements from a
PCollection
to a sink.A Writer
iobase.Writer.write()
writes and elements to the sink whileiobase.Writer.close()
is called after all elements in the bundle have been written.See
iobase.Sink
for more detailed documentation about the process of writing to a sink.
-
class
apache_beam.io.iobase.
Read
(source: apache_beam.io.iobase.SourceBase)[source]¶ Bases:
apache_beam.transforms.ptransform.PTransform
A transform that reads a PCollection.
Initializes a Read transform.
Parameters: source – Data source to read from. -
to_runner_api_parameter
(context: apache_beam.runners.pipeline_context.PipelineContext) → Tuple[str, Any][source]¶
-
static
from_runner_api_parameter
(transform: org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.PTransform, payload: Union[org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.ReadPayload, org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.PubSubReadPayload], context: apache_beam.runners.pipeline_context.PipelineContext) → apache_beam.io.iobase.Read[source]¶
-
class
PipelineContext
(proto: Union[org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.Components, org.apache.beam.model.fn_execution.v1.beam_fn_api_pb2.ProcessBundleDescriptor, None] = None, component_id_map: Optional[apache_beam.pipeline.ComponentIdMap] = None, default_environment: Optional[apache_beam.transforms.environments.Environment] = None, use_fake_coders: bool = False, iterable_state_read: Optional[Callable[[bytes, CoderImpl], Iterable[T_co]]] = None, iterable_state_write: Optional[Callable[[Iterable[T_co], CoderImpl], bytes]] = None, namespace: str = 'ref', requirements: Iterable[str] = ())¶ Bases:
object
For internal use only; no backwards-compatibility guarantees.
Used for accessing and constructing the referenced objects of a Pipeline.
-
add_requirement
(requirement: str) → None¶
-
coder_id_from_element_type
(element_type: Any, requires_deterministic_key_coder: Optional[str] = None) → str¶
-
default_environment_id
() → str¶
-
deterministic_coder
(coder: apache_beam.coders.coders.Coder, msg: str) → apache_beam.coders.coders.Coder¶
-
element_type_from_coder_id
(coder_id: str) → Any¶
-
static
from_runner_api
(proto: org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.Components) → apache_beam.runners.pipeline_context.PipelineContext¶
-
get_environment_id_for_resource_hints
(hints: Dict[str, bytes]) → str¶ Returns an environment id that has necessary resource hints.
-
requirements
() → FrozenSet[str]¶
-
to_runner_api
() → org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.Components¶
-
-
-
class
apache_beam.io.iobase.
Write
(sink)[source]¶ Bases:
apache_beam.transforms.ptransform.PTransform
A
PTransform
that writes to a sink.A sink should inherit
iobase.Sink
. Such implementations are handled using a composite transform that consists of threeParDo``s - (1) a ``ParDo
performing a global initialization (2) aParDo
performing a parallel write and (3) aParDo
performing a global finalization. In the case of an emptyPCollection
, only the global initialization and finalization will be performed. Currently only batch workflows support custom sinks.Example usage:
pcollection | beam.io.Write(MySink())
This returns a
pvalue.PValue
object that represents the end of the Pipeline.The sink argument may also be a full PTransform, in which case it will be applied directly. This allows composite sink-like transforms (e.g. a sink with some pre-processing DoFns) to be used the same as all other sinks.
This transform also supports sinks that inherit
iobase.NativeSink
. These are sinks that are implemented natively by the Dataflow service and hence should not be updated by users. These sinks are processed using a Dataflow native write transform.Initializes a Write transform.
Parameters: sink – Data sink to write to. -
to_runner_api_parameter
(context: apache_beam.runners.pipeline_context.PipelineContext) → Tuple[str, Any][source]¶
-
static
from_runner_api_parameter
(ptransform: Any, payload: org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.PubSubWritePayload, unused_context: apache_beam.runners.pipeline_context.PipelineContext) → apache_beam.io.iobase.Write[source]¶
-
class
PipelineContext
(proto: Union[org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.Components, org.apache.beam.model.fn_execution.v1.beam_fn_api_pb2.ProcessBundleDescriptor, None] = None, component_id_map: Optional[apache_beam.pipeline.ComponentIdMap] = None, default_environment: Optional[apache_beam.transforms.environments.Environment] = None, use_fake_coders: bool = False, iterable_state_read: Optional[Callable[[bytes, CoderImpl], Iterable[T_co]]] = None, iterable_state_write: Optional[Callable[[Iterable[T_co], CoderImpl], bytes]] = None, namespace: str = 'ref', requirements: Iterable[str] = ())¶ Bases:
object
For internal use only; no backwards-compatibility guarantees.
Used for accessing and constructing the referenced objects of a Pipeline.
-
add_requirement
(requirement: str) → None¶
-
coder_id_from_element_type
(element_type: Any, requires_deterministic_key_coder: Optional[str] = None) → str¶
-
default_environment_id
() → str¶
-
deterministic_coder
(coder: apache_beam.coders.coders.Coder, msg: str) → apache_beam.coders.coders.Coder¶
-
element_type_from_coder_id
(coder_id: str) → Any¶
-
static
from_runner_api
(proto: org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.Components) → apache_beam.runners.pipeline_context.PipelineContext¶
-
get_environment_id_for_resource_hints
(hints: Dict[str, bytes]) → str¶ Returns an environment id that has necessary resource hints.
-
requirements
() → FrozenSet[str]¶
-
to_runner_api
() → org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.Components¶
-
-
-
class
apache_beam.io.iobase.
RestrictionTracker
[source]¶ Bases:
object
Manages access to a restriction.
Keeps track of the restrictions claimed part for a Splittable DoFn.
The restriction may be modified by different threads, however the system will ensure sufficient locking such that no methods on the restriction tracker will be called concurrently.
See following documents for more details. * https://s.apache.org/splittable-do-fn * https://s.apache.org/splittable-do-fn-python-sdk
-
current_restriction
()[source]¶ Returns the current restriction.
Returns a restriction accurately describing the full range of work the current
DoFn.process()
call will do, including already completed work.The current restriction returned by method may be updated dynamically due to due to concurrent invocation of other methods of the
RestrictionTracker
, For example,split()
.This API is required to be implemented.
Returns: a restriction object.
-
current_progress
() → apache_beam.io.iobase.RestrictionProgress[source]¶ Returns a RestrictionProgress object representing the current progress.
This API is recommended to be implemented. The runner can do a better job at parallel processing with better progress signals.
-
check_done
()[source]¶ Checks whether the restriction has been fully processed.
Called by the SDK harness after iterator returned by
DoFn.process()
has been fully read.This method must raise a ValueError if there is still any unclaimed work remaining in the restriction when this method is invoked. Exception raised must have an informative error message.
This API is required to be implemented in order to make sure no data loss during SDK processing.
Returns:
True
if current restriction has been fully processed. :raises:ValueError
– if there is still any unclaimed work remaining.
-
try_split
(fraction_of_remainder)[source]¶ Splits current restriction based on fraction_of_remainder.
If splitting the current restriction is possible, the current restriction is split into a primary and residual restriction pair. This invocation updates the
current_restriction()
to be the primary restriction effectively having the currentDoFn.process()
execution responsible for performing the work that the primary restriction represents. The residual restriction will be executed in a separateDoFn.process()
invocation (likely in a different process). The work performed by executing the primary and residual restrictions as separateDoFn.process()
invocations MUST be equivalent to the work performed as if this split never occurred.The
fraction_of_remainder
should be used in a best effort manner to choose a primary and residual restriction based upon the fraction of the remaining work that the currentDoFn.process()
invocation is responsible for. For example, if aDoFn.process()
was reading a file with a restriction representing the offset range [100, 200) and has processed up to offset 130 with a fraction_of_remainder of 0.7, the primary and residual restrictions returned would be [100, 179), [179, 200) (note: current_offset + fraction_of_remainder * remaining_work = 130 + 0.7 * 70 = 179).fraction_of_remainder
= 0 means a checkpoint is required.The API is recommended to be implemented for batch pipeline given that it is very important for pipeline scaling and end to end pipeline execution.
The API is required to be implemented for a streaming pipeline.
Parameters: fraction_of_remainder – A hint as to the fraction of work the primary restriction should represent based upon the current known remaining amount of work. Returns: (primary_restriction, residual_restriction) if a split was possible, otherwise returns None
.
-
try_claim
(position)[source]¶ Attempts to claim the block of work in the current restriction identified by the given position. Each claimed position MUST be a valid split point.
If this succeeds, the DoFn MUST execute the entire block of work. If it fails, the
DoFn.process()
MUST returnNone
without performing any additional work or emitting output (note that emitting output or performing work fromDoFn.process()
is also not allowed before the first call of this method).The API is required to be implemented.
Parameters: position – current position that wants to be claimed. Returns:
True
if the position can be claimed as current_position. Otherwise, returnsFalse
.
-
is_bounded
()[source]¶ Returns whether the amount of work represented by the current restriction is bounded.
The boundedness of the restriction is used to determine the default behavior of how to truncate restrictions when a pipeline is being drained. # pylint: disable=line-too-long If the restriction is bounded, then the entire restriction will be processed otherwise the restriction will be processed till a checkpoint is possible.
The API is required to be implemented.
Returns:
True
if the restriction represents a finite amount of work. Otherwise, returnsFalse
.
-
-
class
apache_beam.io.iobase.
WatermarkEstimator
[source]¶ Bases:
object
A WatermarkEstimator which is used for estimating output_watermark based on the timestamp of output records or manual modifications. Please refer to
watermark_estiamtors
for commonly used watermark estimators.The base class provides common APIs that are called by the framework, which are also accessible inside a DoFn.process() body. Derived watermark estimator should implement all APIs listed below. Additional methods can be implemented and will be available when invoked within a DoFn.
Internal state must not be updated asynchronously.
-
get_estimator_state
()[source]¶ Get current state of the WatermarkEstimator instance, which can be used to recreate the WatermarkEstimator when processing the restriction. See WatermarkEstimatorProvider.create_watermark_estimator.
-