apache_beam.io.fileio module¶
PTransforms
for manipulating files in Apache Beam.
Provides reading PTransform
s, MatchFiles
,
MatchAll
, that produces a PCollection
of records representing a file
and its metadata; and ReadMatches
, which takes in a PCollection
of file
metadata records, and produces a PCollection
of ReadableFile
objects.
These transforms currently do not support splitting by themselves.
Writing to Files¶
The transforms in this file include WriteToFiles
, which allows you to write
a beam.PCollection
to files, and gives you many options to customize how to
do this.
The WriteToFiles
transform supports bounded and unbounded PCollections
(i.e. it can be used both batch and streaming pipelines). For streaming
pipelines, it currently does not have support for multiple trigger firings
on the same window.
File Naming¶
One of the parameters received by WriteToFiles
is a function specifying how
to name the files that are written. This is a function that takes in the
following parameters:
- window
- pane
- shard_index
- total_shards
- compression
- destination
It should return a file name that is unique for a combination of these parameters.
The default naming strategy is to name files in the format $prefix-$start-$end-$pane-$shard-of-$numShards$suffix$compressionSuffix, where:
- $prefix is, by default, “output”.
- $start and $end are the boundaries of the window for the data being written. These are omitted if we’re using the Global window.
- $pane is the index for the number of firing for a window.
- $shard and $numShards are the current shard number, and the total number of shards for this window firing.
- $suffix is, by default, an empty string, but it can be set by the user via
default_file_naming
.
Dynamic Destinations¶
If the elements in the input beam.PCollection
can be partitioned into groups
that should be treated differently (e.g. some events are to be stored as CSV,
while some others are to be stored as Avro files), it is possible to do this
by passing a destination parameter to WriteToFiles
. Something like the
following:
my_pcollection | beam.io.fileio.WriteToFiles(
path='/my/file/path',
destination=lambda record: 'avro' if record['type'] == 'A' else 'csv',
sink=lambda dest: AvroSink() if dest == 'avro' else CsvSink(),
file_naming=beam.io.fileio.destination_prefix_naming())
In this transform, depending on the type of a record, it will be written down to a destination named ‘avro’, or ‘csv’. The value returned by the destination call is then passed to the sink call, to determine what sort of sink will be used for each destination. The return type of the destination parameter can be anything, as long as elements can be grouped by it.
-
class
apache_beam.io.fileio.
EmptyMatchTreatment
[source]¶ Bases:
object
How to treat empty matches in
MatchAll
andMatchFiles
transforms.If empty matches are disallowed, an error will be thrown if a pattern does not match any files.
-
ALLOW
= 'ALLOW'¶
-
DISALLOW
= 'DISALLOW'¶
-
ALLOW_IF_WILDCARD
= 'ALLOW_IF_WILDCARD'¶
-
-
class
apache_beam.io.fileio.
MatchFiles
(file_pattern: str, empty_match_treatment='ALLOW_IF_WILDCARD')[source]¶ Bases:
apache_beam.transforms.ptransform.PTransform
Matches a file pattern using
FileSystems.match
.This
PTransform
returns aPCollection
of matching files in the form ofFileMetadata
objects.
-
class
apache_beam.io.fileio.
MatchAll
(empty_match_treatment='ALLOW')[source]¶ Bases:
apache_beam.transforms.ptransform.PTransform
Matches file patterns from the input PCollection via
FileSystems.match
.This
PTransform
returns aPCollection
of matching files in the form ofFileMetadata
objects.
-
class
apache_beam.io.fileio.
ReadableFile
(metadata, compression=None)[source]¶ Bases:
object
A utility class for accessing files.
-
class
apache_beam.io.fileio.
MatchContinuously
(file_pattern, interval=360.0, has_deduplication=True, start_timestamp=Timestamp(1665645575.385199), stop_timestamp=Timestamp(9223372036854.775000), match_updated_files=False, apply_windowing=False)[source]¶ Bases:
apache_beam.transforms.ptransform.PTransform
Checks for new files for a given pattern every interval.
This
PTransform
returns aPCollection
of matching files in the form ofFileMetadata
objects.MatchContinuously is experimental. No backwards-compatibility guarantees.
Initializes a MatchContinuously transform.
Parameters: - file_pattern – The file path to read from.
- interval – Interval at which to check for files in seconds.
- has_deduplication – Whether files already read are discarded or not.
- start_timestamp – Timestamp for start file checking.
- stop_timestamp – Timestamp after which no more files will be checked.
- match_updated_files – (When has_deduplication is set to True) whether match file with timestamp changes.
- apply_windowing – Whether each element should be assigned to individual window. If false, all elements will reside in global window.
-
class
apache_beam.io.fileio.
ReadMatches
(compression=None, skip_directories=True)[source]¶ Bases:
apache_beam.transforms.ptransform.PTransform
Converts each result of MatchFiles() or MatchAll() to a ReadableFile.
This helps read in a file’s contents or obtain a file descriptor.
-
class
apache_beam.io.fileio.
WriteToFiles
(path, file_naming=None, destination=None, temp_directory=None, sink=None, shards=None, output_fn=None, max_writers_per_bundle=20)[source]¶ Bases:
apache_beam.transforms.ptransform.PTransform
Write the incoming PCollection to a set of output files.
The incoming
PCollection
may be bounded or unbounded.Note: For unbounded
PCollection
s, this transform does not support multiple firings per Window (due to the fact that files are named only by their destination, and window, at the moment).WriteToFiles is experimental. No backwards-compatibility guarantees.
Initializes a WriteToFiles transform.
Parameters: - path (str, ValueProvider) – The directory to write files into.
- file_naming (callable) – A callable that takes in a window, pane, shard_index, total_shards and compression; and returns a file name.
- destination (callable) – If this argument is provided, the sink parameter must also be a callable.
- temp_directory (str, ValueProvider) – To ensure atomicity in the transform, the output is written into temporary files, which are written to a directory that is meant to be temporary as well. Once the whole output has been written, the files are moved into their final destination, and given their final names. By default, the temporary directory will be within the temp_location of your pipeline.
- sink (callable, FileSink) – The sink to use to write
into a file. It should implement the methods of a
FileSink
. Pass a class signature or an instance of FileSink to this parameter. If none is provided, aTextSink
is used. - shards (int) – The number of shards per destination and trigger firing.
- max_writers_per_bundle (int) – The number of writers that can be open concurrently in a single worker that’s processing one bundle.
-
MAX_NUM_WRITERS_PER_BUNDLE
= 20¶
-
DEFAULT_SHARDING
= 5¶