#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""A framework for developing sources for new file types.
To create a source for a new file type a sub-class of :class:`FileBasedSource`
should be created. Sub-classes of :class:`FileBasedSource` must implement the
method :meth:`FileBasedSource.read_records()`. Please read the documentation of
that method for more details.
For an example implementation of :class:`FileBasedSource` see
:class:`~apache_beam.io._AvroSource`.
"""
# pytype: skip-file
from typing import Callable
from typing import Iterable
from typing import Tuple
from typing import Union
from apache_beam.internal import pickler
from apache_beam.io import concat_source
from apache_beam.io import iobase
from apache_beam.io import range_trackers
from apache_beam.io.filesystem import CompressionTypes
from apache_beam.io.filesystem import FileMetadata
from apache_beam.io.filesystem import FileSystem
from apache_beam.io.filesystems import FileSystems
from apache_beam.io.restriction_trackers import OffsetRange
from apache_beam.options.value_provider import StaticValueProvider
from apache_beam.options.value_provider import ValueProvider
from apache_beam.options.value_provider import check_accessible
from apache_beam.transforms.core import DoFn
from apache_beam.transforms.core import ParDo
from apache_beam.transforms.core import PTransform
from apache_beam.transforms.display import DisplayDataItem
from apache_beam.transforms.util import Reshuffle
MAX_NUM_THREADS_FOR_SIZE_ESTIMATION = 25
__all__ = ['FileBasedSource']
[docs]
class FileBasedSource(iobase.BoundedSource):
"""A :class:`~apache_beam.io.iobase.BoundedSource` for reading a file glob of
a given type."""
MIN_NUMBER_OF_FILES_TO_STAT = 100
MIN_FRACTION_OF_FILES_TO_STAT = 0.01
def __init__(
self,
file_pattern,
min_bundle_size=0,
compression_type=CompressionTypes.AUTO,
splittable=True,
validate=True):
"""Initializes :class:`FileBasedSource`.
Args:
file_pattern (str): the file glob to read a string or a
:class:`~apache_beam.options.value_provider.ValueProvider`
(placeholder to inject a runtime value).
min_bundle_size (int): minimum size of bundles that should be generated
when performing initial splitting on this source.
compression_type (str): Used to handle compressed output files.
Typical value is :attr:`CompressionTypes.AUTO
<apache_beam.io.filesystem.CompressionTypes.AUTO>`,
in which case the final file path's extension will be used to detect
the compression.
splittable (bool): whether :class:`FileBasedSource` should try to
logically split a single file into data ranges so that different parts
of the same file can be read in parallel. If set to :data:`False`,
:class:`FileBasedSource` will prevent both initial and dynamic splitting
of sources for single files. File patterns that represent multiple files
may still get split into sources for individual files. Even if set to
:data:`True` by the user, :class:`FileBasedSource` may choose to not
split the file, for example, for compressed files where currently it is
not possible to efficiently read a data range without decompressing the
whole file.
validate (bool): Boolean flag to verify that the files exist during the
pipeline creation time.
Raises:
TypeError: when **compression_type** is not valid or if
**file_pattern** is not a :class:`str` or a
:class:`~apache_beam.options.value_provider.ValueProvider`.
ValueError: when compression and splittable files are
specified.
IOError: when the file pattern specified yields an empty
result.
"""
if not isinstance(file_pattern, (str, ValueProvider)):
raise TypeError(
'%s: file_pattern must be of type string'
' or ValueProvider; got %r instead' %
(self.__class__.__name__, file_pattern))
if isinstance(file_pattern, str):
file_pattern = StaticValueProvider(str, file_pattern)
self._pattern = file_pattern
self._concat_source = None
self._min_bundle_size = min_bundle_size
if not CompressionTypes.is_valid_compression_type(compression_type):
raise TypeError(
'compression_type must be CompressionType object but '
'was %s' % type(compression_type))
self._compression_type = compression_type
self._splittable = splittable
if validate and file_pattern.is_accessible():
self._validate()
[docs]
def display_data(self):
return {
'file_pattern': DisplayDataItem(
str(self._pattern), label="File Pattern"),
'compression': DisplayDataItem(
str(self._compression_type), label='Compression Type')
}
@check_accessible(['_pattern'])
def _get_concat_source(self) -> concat_source.ConcatSource:
if self._concat_source is None:
pattern = self._pattern.get()
single_file_sources = []
match_result = FileSystems.match([pattern])[0]
files_metadata = match_result.metadata_list
# We create a reference for FileBasedSource that will be serialized along
# with each _SingleFileSource. To prevent this FileBasedSource from having
# a reference to ConcatSource (resulting in quadratic space complexity)
# we clone it here.
file_based_source_ref = pickler.loads(pickler.dumps(self))
for file_metadata in files_metadata:
file_name = file_metadata.path
file_size = file_metadata.size_in_bytes
if file_size == 0:
continue # Ignoring empty file.
# We determine splittability of this specific file.
splittable = (
self.splittable and _determine_splittability_from_compression_type(
file_name, self._compression_type))
single_file_source = _SingleFileSource(
file_based_source_ref,
file_name,
0,
file_size,
min_bundle_size=self._min_bundle_size,
splittable=splittable)
single_file_sources.append(single_file_source)
self._report_source_lineage(files_metadata)
self._concat_source = concat_source.ConcatSource(single_file_sources)
return self._concat_source
def _report_source_lineage(self, files_metadata):
"""
Report source Lineage. depend on the number of files, report full file
name, only dir, or only top level
"""
if len(files_metadata) <= 100:
for file_metadata in files_metadata:
FileSystems.report_source_lineage(file_metadata.path)
else:
size_track = set()
for file_metadata in files_metadata:
if len(size_track) >= 100:
FileSystems.report_source_lineage(
file_metadata.path, level=FileSystem.LineageLevel.TOP_LEVEL)
return
try:
base, _ = FileSystems.split(file_metadata.path)
except ValueError:
pass
else:
size_track.add(base)
for base in size_track:
FileSystems.report_source_lineage(base)
[docs]
def open_file(self, file_name):
return FileSystems.open(
file_name,
'application/octet-stream',
compression_type=self._compression_type)
@check_accessible(['_pattern'])
def _validate(self):
"""Validate if there are actual files in the specified glob pattern
"""
pattern = self._pattern.get()
# Limit the responses as we only want to check if something exists
match_result = FileSystems.match([pattern], limits=[1])[0]
if len(match_result.metadata_list) <= 0:
raise IOError('No files found based on the file pattern %s' % pattern)
[docs]
def split(
self, desired_bundle_size=None, start_position=None, stop_position=None):
return self._get_concat_source().split(
desired_bundle_size=desired_bundle_size,
start_position=start_position,
stop_position=stop_position)
[docs]
def estimate_size(self):
return self._get_concat_source().estimate_size()
[docs]
def read(self, range_tracker):
return self._get_concat_source().read(range_tracker)
[docs]
def get_range_tracker(self, start_position, stop_position):
return self._get_concat_source().get_range_tracker(
start_position, stop_position)
[docs]
def read_records(self, file_name, offset_range_tracker):
"""Returns a generator of records created by reading file 'file_name'.
Args:
file_name: a ``string`` that gives the name of the file to be read. Method
``FileBasedSource.open_file()`` must be used to open the file
and create a seekable file object.
offset_range_tracker: a object of type ``OffsetRangeTracker``. This
defines the byte range of the file that should be
read. See documentation in
``iobase.BoundedSource.read()`` for more information
on reading records while complying to the range
defined by a given ``RangeTracker``.
Returns:
an iterator that gives the records read from the given file.
"""
raise NotImplementedError
@property
def splittable(self):
return self._splittable
def _determine_splittability_from_compression_type(file_path, compression_type):
if compression_type == CompressionTypes.AUTO:
compression_type = CompressionTypes.detect_compression_type(file_path)
return compression_type == CompressionTypes.UNCOMPRESSED
class _SingleFileSource(iobase.BoundedSource):
"""Denotes a source for a specific file type."""
def __init__(
self,
file_based_source,
file_name,
start_offset,
stop_offset,
min_bundle_size=0,
splittable=True):
if not isinstance(start_offset, int):
raise TypeError(
'start_offset must be a number. Received: %r' % start_offset)
if stop_offset != range_trackers.OffsetRangeTracker.OFFSET_INFINITY:
if not isinstance(stop_offset, int):
raise TypeError(
'stop_offset must be a number. Received: %r' % stop_offset)
if start_offset >= stop_offset:
raise ValueError(
'start_offset must be smaller than stop_offset. Received %d and %d '
'for start and stop offsets respectively' %
(start_offset, stop_offset))
self._file_name = file_name
self._is_gcs_file = file_name.startswith('gs://') if file_name else False
self._start_offset = start_offset
self._stop_offset = stop_offset
self._min_bundle_size = min_bundle_size
self._file_based_source = file_based_source
self._splittable = splittable
def split(self, desired_bundle_size, start_offset=None, stop_offset=None):
if start_offset is None:
start_offset = self._start_offset
if stop_offset is None:
stop_offset = self._stop_offset
if self._splittable:
splits = OffsetRange(start_offset, stop_offset).split(
desired_bundle_size, self._min_bundle_size)
for split in splits:
yield iobase.SourceBundle(
split.stop - split.start,
_SingleFileSource(
# Copying this so that each sub-source gets a fresh instance.
pickler.loads(pickler.dumps(self._file_based_source)),
self._file_name,
split.start,
split.stop,
min_bundle_size=self._min_bundle_size,
splittable=self._splittable),
split.start,
split.stop)
else:
# Returning a single sub-source with end offset set to OFFSET_INFINITY (so
# that all data of the source gets read) since this source is
# unsplittable. Choosing size of the file as end offset will be wrong for
# certain unsplittable source, e.g., compressed sources.
yield iobase.SourceBundle(
stop_offset - start_offset,
_SingleFileSource(
self._file_based_source,
self._file_name,
start_offset,
range_trackers.OffsetRangeTracker.OFFSET_INFINITY,
min_bundle_size=self._min_bundle_size,
splittable=self._splittable),
start_offset,
range_trackers.OffsetRangeTracker.OFFSET_INFINITY)
def estimate_size(self):
return self._stop_offset - self._start_offset
def get_range_tracker(self, start_position, stop_position):
if start_position is None:
start_position = self._start_offset
if stop_position is None:
# If file is unsplittable we choose OFFSET_INFINITY as the default end
# offset so that all data of the source gets read. Choosing size of the
# file as end offset will be wrong for certain unsplittable source, for
# e.g., compressed sources.
stop_position = (
self._stop_offset if self._splittable else
range_trackers.OffsetRangeTracker.OFFSET_INFINITY)
range_tracker = range_trackers.OffsetRangeTracker(
start_position, stop_position)
if not self._splittable:
range_tracker = range_trackers.UnsplittableRangeTracker(range_tracker)
return range_tracker
def read(self, range_tracker):
return self._file_based_source.read_records(self._file_name, range_tracker)
def default_output_coder(self):
return self._file_based_source.default_output_coder()
class _ExpandIntoRanges(DoFn):
def __init__(
self, splittable, compression_type, desired_bundle_size, min_bundle_size):
self._desired_bundle_size = desired_bundle_size
self._min_bundle_size = min_bundle_size
self._splittable = splittable
self._compression_type = compression_type
self._size_track = None
def process(self, element: Union[str, FileMetadata], *args,
**kwargs) -> Iterable[Tuple[FileMetadata, OffsetRange]]:
if isinstance(element, FileMetadata):
metadata_list = [element]
else:
match_results = FileSystems.match([element])
metadata_list = match_results[0].metadata_list
for metadata in metadata_list:
self._report_source_lineage(metadata.path)
splittable = (
self._splittable and _determine_splittability_from_compression_type(
metadata.path, self._compression_type))
if splittable:
for split in OffsetRange(0, metadata.size_in_bytes).split(
self._desired_bundle_size, self._min_bundle_size):
yield (metadata, split)
else:
yield (
metadata,
OffsetRange(0, range_trackers.OffsetRangeTracker.OFFSET_INFINITY))
def _report_source_lineage(self, path):
"""
Report source Lineage. Due to the size limit of Beam metrics, report full
file name or only top level depend on the number of files.
* Number of files<=100, report full file paths;
* Otherwise, report top level only.
"""
if self._size_track is None:
self._size_track = set()
elif len(self._size_track) == 0:
FileSystems.report_source_lineage(
path, level=FileSystem.LineageLevel.TOP_LEVEL)
return
self._size_track.add(path)
FileSystems.report_source_lineage(path)
if len(self._size_track) >= 100:
self._size_track.clear()
class _ReadRange(DoFn):
def __init__(
self,
source_from_file: Union[str, iobase.BoundedSource],
with_filename: bool = False) -> None:
self._source_from_file = source_from_file
self._with_filename = with_filename
def process(self, element, *args, **kwargs):
metadata, range = element
source = self._source_from_file(metadata.path)
# Following split() operation has to be performed to create a proper
# _SingleFileSource. Otherwise what we have is a ConcatSource that contains
# a single _SingleFileSource. ConcatSource.read() expects a RangeTracker for
# sub-source range and reads full sub-sources (not byte ranges).
source_list = list(source.split(float('inf')))
# Handle the case of an empty source.
if not source_list:
return
source = source_list[0].source
for record in source.read(range.new_tracker()):
if self._with_filename:
yield (metadata.path, record)
else:
yield record
class ReadAllFiles(PTransform):
"""A Read transform that reads a PCollection of files.
Pipeline authors should not use this directly. This is to be used by Read
PTransform authors who wishes to implement file-based Read transforms that
read a PCollection of files.
"""
def __init__(
self,
splittable: bool,
compression_type,
desired_bundle_size: int,
min_bundle_size: int,
source_from_file: Callable[[str], iobase.BoundedSource],
with_filename: bool = False):
"""
Args:
splittable: If False, files won't be split into sub-ranges. If True,
files may or may not be split into data ranges.
compression_type: A ``CompressionType`` object that specifies the
compression type of the files that will be processed. If
``CompressionType.AUTO``, system will try to automatically
determine the compression type based on the extension of
files.
desired_bundle_size: the desired size of data ranges that should be
generated when splitting a file into data ranges.
min_bundle_size: minimum size of data ranges that should be generated when
splitting a file into data ranges.
source_from_file: a function that produces a ``BoundedSource`` given a
file name. System will use this function to generate
``BoundedSource`` objects for file paths. Note that file
paths passed to this will be for individual files, not
for file patterns even if the ``PCollection`` of files
processed by the transform consist of file patterns.
with_filename: If True, returns a Key Value with the key being the file
name and the value being the actual data. If False, it only returns
the data.
"""
self._splittable = splittable
self._compression_type = compression_type
self._desired_bundle_size = desired_bundle_size
self._min_bundle_size = min_bundle_size
self._source_from_file = source_from_file
self._with_filename = with_filename
# TODO(BEAM-14497) always reshuffle once gbk always trigger works.
self._is_reshuffle = True
def _disable_reshuffle(self):
# TODO(BEAM-14497) Remove this private method once gbk always trigger works.
#
# Currently Reshuffle() holds elements until the stage is completed. When
# ReadRange is needed instantly after match (like read continuously), the
# reshard is temporarily disabled. However, the read then does not scale and
# is deemed experimental.
self._is_reshuffle = False
return self
def expand(self, pvalue):
pvalue = (
pvalue
| 'ExpandIntoRanges' >> ParDo(
_ExpandIntoRanges(
self._splittable,
self._compression_type,
self._desired_bundle_size,
self._min_bundle_size)))
if self._is_reshuffle:
pvalue = pvalue | 'Reshard' >> Reshuffle()
return (
pvalue
| 'ReadRange' >> ParDo(
_ReadRange(
self._source_from_file, with_filename=self._with_filename)))