This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 7bda6f6 [FLINK-13842][docs] Improve Javadocs and web documentation of the StreamingFileSink 7bda6f6 is described below commit 7bda6f61c9ce0db80f2b1d4c8e7fd6cad9b0eaf9 Author: Gyula Fora <gyf...@apache.org> AuthorDate: Thu Sep 5 16:04:25 2019 +0200 [FLINK-13842][docs] Improve Javadocs and web documentation of the StreamingFileSink Closes #9530 --- docs/dev/connectors/streamfile_sink.md | 339 +++++++++++++++++---- docs/fig/streamfilesink_bucketing.png | Bin 0 -> 100661 bytes .../sink/filesystem/StreamingFileSink.java | 25 ++ .../rollingpolicies/DefaultRollingPolicy.java | 6 + 4 files changed, 316 insertions(+), 54 deletions(-) diff --git a/docs/dev/connectors/streamfile_sink.md b/docs/dev/connectors/streamfile_sink.md index 9017447..c59e646 100644 --- a/docs/dev/connectors/streamfile_sink.md +++ b/docs/dev/connectors/streamfile_sink.md @@ -23,30 +23,140 @@ specific language governing permissions and limitations under the License. --> +* This will be replaced by the TOC +{:toc} + This connector provides a Sink that writes partitioned files to filesystems supported by the [Flink `FileSystem` abstraction]({{ site.baseurl}}/ops/filesystems/index.html). -Since in streaming the input is potentially infinite, the streaming file sink writes data -into buckets. The bucketing behaviour is configurable but a useful default is time-based -bucketing where we start writing a new bucket every hour and thus get -individual files that each contain a part of the infinite output stream. +In order to handle unbounded data streams, the streaming file sink writes incoming data +into buckets. The bucketing behaviour is fully configurable with a default time-based +bucketing where we start writing a new bucket every hour and thus get files that correspond to +records received during certain time intervals from the stream. + +The bucket directories themselves contain several part files with the actual output data, with at least +one for each subtask of the sink that has received data for the bucket. Additional part files will be created according to the configurable +rolling policy. The default policy rolls files based on size, a timeout that specifies the maximum duration for which a file can be open, and a maximum inactivity timeout after which the file is closed. + + <div class="alert alert-info"> + <b>IMPORTANT:</b> Checkpointing needs to be enabled when using the StreamingFileSink. Part files can only be finalized + on successful checkpoints. If checkpointing is disabled part files will forever stay in `in-progress` or `pending` state + and cannot be safely read by downstream systems. + </div> + + <img src="{{ site.baseurl }}/fig/streamfilesink_bucketing.png" class="center" style="width: 100%;" /> + +### Bucket Assignment + +The bucketing logic defines how the data will be structured into subdirectories inside the base output directory. + +Both row and bulk formats use the [DateTimeBucketAssigner]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssigner.html) as the default assigner. +By default the DateTimeBucketAssigner creates hourly buckets based on the system default timezone +with the following format: `yyyy-MM-dd--HH`. Both the date format (i.e. bucket size) and timezone can be +configured manually. + +We can specify a custom [BucketAssigner]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketAssigner.html) by calling `.withBucketAssigner(assigner)` on the format builders. + +Flink comes with two built in BucketAssigners: + + - [DateTimeBucketAssigner]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssigner.html) : Default time based assigner + - [BasePathBucketAssigner]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/BasePathBucketAssigner.html) : Assigner that stores all part files in the base path (single global bucket) + +### Rolling Policy + +The [RollingPolicy]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicy.html) defines when a given in-progress part file will be closed and moved to the pending and later to finished state. +In combination with the checkpointing interval (pending files become finished on the next checkpoint) this controls how quickly +part files become available for downstream readers and also the size and number of these parts. + +Flink comes with two built-in RollingPolicies: + + - [DefaultRollingPolicy]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.html) + - [OnCheckpointRollingPolicy]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/OnCheckpointRollingPolicy.html) + +### Part file lifecycle + +In order to use the output of the StreamingFileSink in downstream systems, we need to understand the naming and lifecycle of the output files produced. + +Part files can be in one of three states: + 1. **In-progress** : The part file that is currently being written to is in-progress + 2. **Pending** : Once a part file is closed for writing it becomes pending + 3. **Finished** : On successful checkpoints pending files become finished + +Only finished files are safe to read by downstream systems as those are guaranteed to not be modified later. Finished files can be distinguished by their naming scheme only. + +File naming schemes: + - **In-progress / Pending**: `part-subtaskIndex-partFileIndex.inprogress.uid` + - **Finished:** `part-subtaskIndex-partFileIndex` + +Part file indexes are strictly increasing for any given subtask (in the order they were created). However these indexes are not always sequential. When the job restarts, the next part index for all subtask will be the `max part index + 1`. + +Each writer subtask will have a single in-progress part file at any given time for every active bucket, but there can be several pending and finished files. + +**Part file example** + +To better understand the lifecycle of these files let's look at a simple example with 2 sink subtasks: + +``` +└── 2019-08-25--12 + ├── part-0-0.inprogress.bd053eb0-5ecf-4c85-8433-9eff486ac334 + └── part-1-0.inprogress.ea65a428-a1d0-4a0b-bbc5-7a436a75e575 +``` + +When the part file `part-1-0` is rolled (let's say it becomes too large), it becomes pending but it is not renamed. The sink then opens a new part file: `part-1-1`: + +``` +└── 2019-08-25--12 + ├── part-0-0.inprogress.bd053eb0-5ecf-4c85-8433-9eff486ac334 + ├── part-1-0.inprogress.ea65a428-a1d0-4a0b-bbc5-7a436a75e575 + └── part-1-1.inprogress.bc279efe-b16f-47d8-b828-00ef6e2fbd11 +``` -Within a bucket, we further split the output into smaller part files based on a -rolling policy. This is useful to prevent individual bucket files from getting -too big. This is also configurable but the default policy rolls files based on -file size and a timeout, *i.e* if no new data was written to a part file. +As `part-1-0` is now pending completion, after the next successful checkpoint, it is finalized: -The `StreamingFileSink` supports both row-wise encoding formats and -bulk-encoding formats, such as [Apache Parquet](http://parquet.apache.org). +``` +└── 2019-08-25--12 + ├── part-0-0.inprogress.bd053eb0-5ecf-4c85-8433-9eff486ac334 + ├── part-1-0 + └── part-1-1.inprogress.bc279efe-b16f-47d8-b828-00ef6e2fbd11 +``` -#### Using Row-encoded Output Formats +New buckets are created as dictated by the bucketing policy, and this doesn't affect currently in-progress files: -The only required configuration are the base path where we want to output our -data and an -[Encoder]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/api/common/serialization/Encoder.html) -that is used for serializing records to the `OutputStream` for each file. +``` +└── 2019-08-25--12 + ├── part-0-0.inprogress.bd053eb0-5ecf-4c85-8433-9eff486ac334 + ├── part-1-0 + └── part-1-1.inprogress.bc279efe-b16f-47d8-b828-00ef6e2fbd11 +└── 2019-08-25--13 + └── part-0-2.inprogress.2b475fec-1482-4dea-9946-eb4353b475f1 +``` -Basic usage thus looks like this: +Old buckets can still receive new records as the bucketing policy is evaluated on a per-record basis. + +## File Formats + +The `StreamingFileSink` supports both row-wise and bulk encoding formats, such as [Apache Parquet](http://parquet.apache.org). +These two variants come with their respective builders that can be created with the following static methods: + + - Row-encoded sink: `StreamingFileSink.forRowFormat(basePath, rowEncoder)` + - Bulk-encoded sink: `StreamingFileSink.forBulkFormat(basePath, bulkWriterFactory)` + +When creating either a row or a bulk encoded sink we have to specify the base path where the buckets will be +stored and the encoding logic for our data. + +Please check out the JavaDoc for [StreamingFileSink]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.html) for all the configuration options +and more documentation about the implementation of the different data formats. + +### Row-encoded Formats + +Row-encoded formats need to specify an [Encoder]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/api/common/serialization/Encoder.html) that is used for serializing individual rows to the `OutputStream` of the in-progress part files. + +In addition to the bucket assigner the [RowFormatBuilder]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.RowFormatBuilder.html) allows the user to specify: + + - Custom [RollingPolicy]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicy.html) : Rolling polciy to override the DefaultRollingPolicy + - bucketCheckInterval (default = 1 min) : Millisecond interval for checking time based rolling policies + +Basic usage for writing String elements thus looks like this: <div class="codetabs" markdown="1"> @@ -55,11 +165,18 @@ Basic usage thus looks like this: import org.apache.flink.api.common.serialization.SimpleStringEncoder; import org.apache.flink.core.fs.Path; import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink; +import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy; DataStream<String> input = ...; final StreamingFileSink<String> sink = StreamingFileSink - .forRowFormat(new Path(outputPath), new SimpleStringEncoder<>("UTF-8")) + .forRowFormat(new Path(outputPath), new SimpleStringEncoder<>("UTF-8")) + .withRollingPolicy( + DefaultRollingPolicy.builder() + .withRolloverInterval(TimeUnit.MINUTES.toMillis(15)) + .withInactivityInterval(TimeUnit.MINUTES.toMillis(5)) + .withMaxPartSize(1024 * 1024 * 1024) + .build()) .build(); input.addSink(sink); @@ -71,55 +188,169 @@ input.addSink(sink); import org.apache.flink.api.common.serialization.SimpleStringEncoder import org.apache.flink.core.fs.Path import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink +import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy val input: DataStream[String] = ... val sink: StreamingFileSink[String] = StreamingFileSink .forRowFormat(new Path(outputPath), new SimpleStringEncoder[String]("UTF-8")) + .withRollingPolicy( + DefaultRollingPolicy.builder() + .withRolloverInterval(TimeUnit.MINUTES.toMillis(15)) + .withInactivityInterval(TimeUnit.MINUTES.toMillis(5)) + .withMaxPartSize(1024 * 1024 * 1024) + .build()) .build() - + input.addSink(sink) {% endhighlight %} </div> </div> -This will create a streaming sink that creates hourly buckets and uses a -default rolling policy. The default bucket assigner is -[DateTimeBucketAssigner]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssigner.html) -and the default rolling policy is -[DefaultRollingPolicy]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.html). -You can specify a custom -[BucketAssigner]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketAssigner.html) -and -[RollingPolicy]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicy.html) -on the sink builder. Please check out the JavaDoc for -[StreamingFileSink]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.html) -for more configuration options and more documentation about the workings and -interactions of bucket assigners and rolling policies. - -#### Using Bulk-encoded Output Formats - -In the above example we used an `Encoder` that can encode or serialize each -record individually. The streaming file sink also supports bulk-encoded output -formats such as [Apache Parquet](http://parquet.apache.org). To use these, -instead of `StreamingFileSink.forRowFormat()` you would use -`StreamingFileSink.forBulkFormat()` and specify a `BulkWriter.Factory`. - -[ParquetAvroWriters]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/formats/parquet/avro/ParquetAvroWriters.html) -has static methods for creating a `BulkWriter.Factory` for various types. - -<div class="alert alert-info"> - <b>IMPORTANT:</b> Bulk-encoding formats can only be combined with the - `OnCheckpointRollingPolicy`, which rolls the in-progress part file on - every checkpoint. +This example creates a simple sink that assigns records to the default one hour time buckets. It also specifies +a rolling policy that rolls the in-progress part file on either of the following 3 conditions: + + - It contains at least 15 minutes worth of data + - It hasn't received new records for the last 5 minutes + - The file size reached 1 GB (after writing the last record) + +### Bulk-encoded Formats + +Bulk-encoded sinks are created similarly to the row-encoded ones but here instead of +specifying an `Encoder` we have to specify [BulkWriter.Factory]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/api/common/serialization/BulkWriter.Factory.html). +The `BulkWriter` logic defines how new elements added, flushed and how the bulk of records +are finalized for further encoding purposes. + +Flink comes with two built-in BulkWriter factories: + + - [ParquetWriterFactory]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/formats/parquet/ParquetWriterFactory.html) + - [SequenceFileWriterFactory]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/formats/sequencefile/SequenceFileWriterFactory.html) + +#### Parquet format + +Flink contains built in convenience methods for creating Parquet writer factories for Avro data. These methods +and their associated documentation can be found in the [ParquetAvroWriters]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/formats/parquet/avro/ParquetAvroWriters.html) class. + +For writing to other Parquet compatible data formats, users need to create the ParquetWriterFactory with a custom implementation of the [ParquetBuilder]({{ site.javadocs_baseurl }}/api/java/org/apache/flink/formats/parquet/ParquetBuilder.html) interface. + +To use the Parquet bulk encoder in your application you need to add the following dependency: + +{% highlight xml %} +<dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-parquet{{ site.scala_version_suffix }}</artifactId> + <version>{{ site.version }}</version> +</dependency> +{% endhighlight %} + +A StreamingFileSink that writes Avro data to Parquet format can be created like this: + +<div class="codetabs" markdown="1"> +<div data-lang="java" markdown="1"> +{% highlight java %} +import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink; +import org.apache.flink.formats.parquet.avro.ParquetAvroWriters; +import org.apache.avro.Schema; + + +Schema schema = ...; +DataStream<GenericRecord> stream = ...; + +final StreamingFileSink<GenericRecord> sink = StreamingFileSink + .forBulkFormat(outputBasePath, ParquetAvroWriters.forGenericRecord(schema)) + .build(); + +input.addSink(sink); + +{% endhighlight %} +</div> +<div data-lang="scala" markdown="1"> +{% highlight scala %} +import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink +import org.apache.flink.formats.parquet.avro.ParquetAvroWriters +import org.apache.avro.Schema + +val schema: Schema = ... +val input: DataStream[GenericRecord] = ... + +val sink: StreamingFileSink[GenericRecord] = StreamingFileSink + .forBulkFormat(outputBasePath, ParquetAvroWriters.forGenericRecord(schema)) + .build() + +input.addSink(sink) + +{% endhighlight %} +</div> </div> -#### Important Considerations for S3 +#### Hadoop SequenceFile format + +To use the SequenceFile bulk encoder in your application you need to add the following dependency: + +{% highlight xml %} +<dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-sequence-file</artifactId> + <version>{{ site.version }}</version> +</dependency> +{% endhighlight %} + +A simple SequenceFile writer can be created like this: + +<div class="codetabs" markdown="1"> +<div data-lang="java" markdown="1"> +{% highlight java %} +import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink; +import org.apache.flink.configuration.GlobalConfiguration; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.SequenceFile; +import org.apache.hadoop.io.Text; + + +DataStream<Tuple2<LongWritable, Text>> input = ...; +Configuration hadoopConf = HadoopUtils.getHadoopConfiguration(GlobalConfiguration.loadConfiguration()); +final StreamingFileSink<Tuple2<LongWritable, Text>> sink = StreamingFileSink + .forBulkFormat( + outputBasePath, + new SequenceFileWriterFactory<>(hadoopConf, LongWritable.class, Text.class)) + .build(); + +input.addSink(sink); + +{% endhighlight %} +</div> +<div data-lang="scala" markdown="1"> +{% highlight scala %} +import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink +import org.apache.flink.configuration.GlobalConfiguration +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.io.LongWritable +import org.apache.hadoop.io.SequenceFile +import org.apache.hadoop.io.Text; + +val input: DataStream[(LongWritable, Text)] = ... +val hadoopConf: Configuration = HadoopUtils.getHadoopConfiguration(GlobalConfiguration.loadConfiguration()) +val sink: StreamingFileSink[(LongWritable, Text)] = StreamingFileSink + .forBulkFormat( + outputBasePath, + new SequenceFileWriterFactory(hadoopConf, LongWritable.class, Text.class)) + .build() + +input.addSink(sink) + +{% endhighlight %} +</div> +</div> + +The SequenceFileWriterFactory supports additional constructor parameters to specify compression settings. + +### Important Considerations for S3 -<span class="label label-danger">Important Note 1</span>: For S3, the `StreamingFileSink` +<span class="label label-danger">Important Note 1</span>: For S3, the `StreamingFileSink` supports only the [Hadoop-based](https://hadoop.apache.org/) FileSystem implementation, not -the implementation based on [Presto](https://prestodb.io/). In case your job uses the +the implementation based on [Presto](https://prestodb.io/). In case your job uses the `StreamingFileSink` to write to S3 but you want to use the Presto-based one for checkpointing, it is advised to use explicitly *"s3a://"* (for Hadoop) as the scheme for the target path of the sink and *"s3p://"* for checkpointing (for Presto). Using *"s3://"* for both the sink @@ -128,10 +359,10 @@ and checkpointing may lead to unpredictable behavior, as both implementations "l <span class="label label-danger">Important Note 2</span>: To guarantee exactly-once semantics while being efficient, the `StreamingFileSink` uses the [Multi-part Upload](https://docs.aws.amazon.com/AmazonS3/latest/dev/mpuoverview.html) feature of S3 (MPU from now on). This feature allows to upload files in independent chunks (thus the "multi-part") -which can be combined into the original file when all the parts of the MPU are successfully uploaded. -For inactive MPUs, S3 supports a bucket lifecycle rule that the user can use to abort multipart uploads -that don't complete within a specified number of days after being initiated. This implies that if you set this rule -aggressively and take a savepoint with some part-files being not fully uploaded, their associated MPUs may time-out +which can be combined into the original file when all the parts of the MPU are successfully uploaded. +For inactive MPUs, S3 supports a bucket lifecycle rule that the user can use to abort multipart uploads +that don't complete within a specified number of days after being initiated. This implies that if you set this rule +aggressively and take a savepoint with some part-files being not fully uploaded, their associated MPUs may time-out before the job is restarted. This will result in your job not being able to restore from that savepoint as the pending part-files are no longer there and Flink will fail with an exception as it tries to fetch them and fails. diff --git a/docs/fig/streamfilesink_bucketing.png b/docs/fig/streamfilesink_bucketing.png new file mode 100644 index 0000000..fe3b226 Binary files /dev/null and b/docs/fig/streamfilesink_bucketing.png differ diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java index 716b4c9..75692dd 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java @@ -224,14 +224,30 @@ public class StreamingFileSink<IN> this.partFileSuffix = Preconditions.checkNotNull(partFileSuffix); } + /** + * Creates a new builder instance with the specified bucket check interval. The interval specifies how often + * time based {@link RollingPolicy}s will be checked/executed for the open buckets. + * @param interval Time interval in milliseconds + * @return A new builder with the check interval set. + */ public StreamingFileSink.RowFormatBuilder<IN, BucketID> withBucketCheckInterval(final long interval) { return new RowFormatBuilder<>(basePath, encoder, bucketAssigner, rollingPolicy, interval, bucketFactory, partFilePrefix, partFileSuffix); } + /** + * Creates a new builder instance with the specified {@link BucketAssigner}. + * @param assigner @{@link BucketAssigner} to be used. + * @return A new builder with the assigner set. + */ public StreamingFileSink.RowFormatBuilder<IN, BucketID> withBucketAssigner(final BucketAssigner<IN, BucketID> assigner) { return new RowFormatBuilder<>(basePath, encoder, Preconditions.checkNotNull(assigner), rollingPolicy, bucketCheckInterval, bucketFactory, partFilePrefix, partFileSuffix); } + /** + * Creates a new builder instance with the specified {@link RollingPolicy} set for the bucketing logic. + * @param policy {@link RollingPolicy} to be applied + * @return A new builder with the check interval set. + */ public StreamingFileSink.RowFormatBuilder<IN, BucketID> withRollingPolicy(final RollingPolicy<IN, BucketID> policy) { return new RowFormatBuilder<>(basePath, encoder, bucketAssigner, Preconditions.checkNotNull(policy), bucketCheckInterval, bucketFactory, partFilePrefix, partFileSuffix); } @@ -314,10 +330,19 @@ public class StreamingFileSink<IN> this.partFileSuffix = Preconditions.checkNotNull(partFileSuffix); } + /** + * Currently bulk formats always use the {@link OnCheckpointRollingPolicy} therefore this settings does + * not have any effect. + */ public StreamingFileSink.BulkFormatBuilder<IN, BucketID> withBucketCheckInterval(long interval) { return new BulkFormatBuilder<>(basePath, writerFactory, bucketAssigner, interval, bucketFactory, partFilePrefix, partFileSuffix); } + /** + * Creates a new builder instance with the specified {@link BucketAssigner}. + * @param assigner @{@link BucketAssigner} to be used. + * @return A new builder with the assigner set. + */ public <ID> StreamingFileSink.BulkFormatBuilder<IN, ID> withBucketAssigner(BucketAssigner<IN, ID> assigner) { return new BulkFormatBuilder<>(basePath, writerFactory, Preconditions.checkNotNull(assigner), bucketCheckInterval, new DefaultBucketFactoryImpl<>(), partFilePrefix, partFileSuffix); } diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java index d9bfbcc..6a12c3c 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java @@ -158,6 +158,9 @@ public final class DefaultRollingPolicy<IN, BucketID> implements RollingPolicy<I /** * Sets the interval of allowed inactivity after which a part file will have to roll. + * The frequency at which this is checked is controlled by the + * {@link org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.RowFormatBuilder#withBucketCheckInterval(long)} + * setting. * @param interval the allowed inactivity interval. */ public DefaultRollingPolicy.PolicyBuilder withInactivityInterval(final long interval) { @@ -167,6 +170,9 @@ public final class DefaultRollingPolicy<IN, BucketID> implements RollingPolicy<I /** * Sets the max time a part file can stay open before having to roll. + * The frequency at which this is checked is controlled by the + * {@link org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.RowFormatBuilder#withBucketCheckInterval(long)} + * setting. * @param interval the desired rollover interval. */ public DefaultRollingPolicy.PolicyBuilder withRolloverInterval(final long interval) {