Hi Steve,
When you call env.readFile(…), internally env creates:
ContinuousFileMonitoringFunction<OUT> monitoringFunction =
new ContinuousFileMonitoringFunction<>(inputFormat, monitoringMode,
getParallelism(), interval);
ContinuousFileReaderOperator<OUT> reader =
new ContinuousFileReaderOperator<>(inputFormat);
SingleOutputStreamOperator<OUT> source = addSource(monitoringFunction,
sourceName)
.transform("Split Reader: " + sourceName, typeInfo, reader);
return new DataStreamSource<>(source);
ContinuousFileMonitoringFunction is RichSource function (not parallel), so it
has parallelism 1.
On each execution (controlled by monitor parameter), it builds recursive tree
of all new file splits which are passed to ContinuousFileReaderOperator.
ContinuousFileReaderOperator in parallel process each filesplit.
From: Steve Whelan <[email protected]>
Date: Tuesday, 11 August 2020 at 04:48
To: user <[email protected]>
Subject: S3 file source parallelism reverting to 1
Hi,
I have an S3 file source being consumed as
FileProcessingMode.PROCESS_CONTINUOUSLY with a parallelism of 3. I can confirm
the parallelism is set by printing it out. However, in the UI, the file source
has a parallelism of 1. I'm not changing it after its being initially set.
DataStream s = env.readFile(
new JsonInputFormat(...),
filePath,
FileProcessingMode.PROCESS_CONTINUOUSLY,
5,
myTypeInformation)
.setParallelism(3);
System.out.println(s.getParallelism()); // prints 3
The DataStreamSource is a parallel operator otherwise `setParallelism(3)` would
throw an `IllegalArgumentException`[1]. The only other thing I do with the
DataStream is register it with the TableEnvironment.
tableEnvironment.registerDataStream("my_table", dataStream);
Is Flink resetting it to 1 for some reason? I'm running v1.9.0.
Thanks,
Steve
[1]
https://github.com/apache/flink/blob/release-1.9.0/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSource.java#L53
