Thanks for the response! That's what I feared was going on.
I consider this a huge shortcoming, particularly because it does not
only affect users with large files like you said. The same happens with
many small files, because file globs are also fused to one worker. The
only way to process files in parallel is to write a PTransform that does
MatchFiles(file_glob) | Reshuffle() | ReadAllFromText(). A simple
ReadFromText(file_glob) would not be run in parallel.
In fact, if you feed multiple textfile sources to your job, not only
will each of these inputs process its files on one worker, but even the
inputs are fused together. So even if you resolved the glob locally and
then added one input for each individual file, all of that would still
run sequentially.
On 09/03/2022 17:14, Jan Lukavský wrote:
Hi Janek,
I think you hit a deficiency in the FlinkRunner's SDF implementation.
AFAIK the runner is unable to do dynamic splitting, which is what you
are probably looking for. What you describe essentially works in the
model, but FlinkRunner does not implement the complete contract to
make use of ability to split a large file to multiple parts and
process them in parallel. I'm afraid there is no simple solution
currently, other than what you described. The dynamic splitting might
be introduced in some future release so that it starts working as you
expect out of the box. This should affect mostly users with few large
files, if you can parallelize on the files itself, then it should work
fine (which is what you observe).
Jan
On 3/9/22 16:44, Janek Bevendorff wrote:
I went through all Flink and Beam documentation I could find to see
if I overlooked something, but I could not get the text input source
to unfuse the file input splits. This creates a huge input
bottleneck, because one worker is busy reading records from a huge
input file while 99 others wait for input and I can only shuffle the
generated records, not the actual file input splits.
To fix it, I wrote a custom PTransform that globs files, optionally
shuffles the file names, generates fixed-size split markers, shuffles
the markers, and then reads the lines from these splits. This works
well, but it feels like a hack, because I think Beam should be doing
that out of the box. My implementation of the file reader is also
much simpler and relies on IOBase.readline(), which keeps the code
short and concise, but also loses a lot of flexibility compared to
the Beam file reader implementation (such as support for custom line
delimiters).
Any other ideas how I can solve this without writing custom PTransforms?
Janek
On 08/03/2022 14:11, Janek Bevendorff wrote:
Hey there,
According to the docs, when using a FileBasedSource or a splittable
DoFn, the runner is free to initiate splits that can be run in
parallel. As far as I can tell, the former is actually happening on
my Apache Flink cluster, but the latter is not. This causes a single
Taskmanager to process all splits of an input text file. Is this
known behaviour and how can I fix this?
I have a pipeline that looks like this (Python SDK):
(pipeline
| 'Read Input File' >> textio.ReadFromText(input_glob,
min_bundle_size=1)
| 'Reshuffle Lines' >> beam.Reshuffle()
| 'Map Records' >> beam.ParDo(map_func))
The input file is a large, uncompressed plaintext file from a shared
drive containing millions of newline-separated data records. I am
running this job with a parallelism of 100, but it is bottlenecked
by a single worker running ReadFromText(). The reshuffling in
between was added to force Beam/Flink to parallelize the processing,
but this has no effect on the preceding stage. Only the following
map operation is being run in parallel. The stage itself is marked
as having a parallelism of 100, but 99 workers finish immediately.
I had the same issue earlier with another input source, in which I
match a bunch of WARC file globs and then iterate over them in a
splittable DoFn. I solved the missing parallelism by adding an
explicit reshuffle in between matching input globs and actually
reading the individual files:
class WarcInput(beam.PTransform):
def expand(self, pcoll):
return pcoll | MatchFiles(self._file_pattern) |
beam.Reshuffle() | beam.ParDo(WarcReader())
This way I can at least achieve parallelism on file level. This
doesn't work with a single splittable input file, of course, for
which I would have to reshuffle somewhere inside of ReadFromText().
Do I really have to write a custom PTransform that generates initial
splits, shuffles them, and then reads from those splits? I consider
this somewhat essential functionality.
Any hints appreciated.
Janek
--
Bauhaus-Universität Weimar
Bauhausstr. 9a, R308
99423 Weimar, Germany
Phone: +49 3643 58 3577
www.webis.de
