[
https://issues.apache.org/jira/browse/FLINK-11838?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17280587#comment-17280587
]
Galen Warren commented on FLINK-11838:
--------------------------------------
Thanks. All of your initial assumptions are correct, though I might add that it
is straightforward to serde a RestorableState<WriteChannel> if Java
serialization is allowed (i.e. the object is guaranteed to implement
Serializable). But I understand that's not preferred.
To answer your questions:
*Does "writing a blob to a temporary location" mean that the user always needs
to configure a temporary location? How is the temporary location cleaned, say
if they're never moved to the committed location?*
I was thinking to manage temporary locations through a couple of Flink options:
* First, a "prefix" option, which would be a string to be applied in front of
the supplied, "final" blob name. This prefix could be default be something like
".inprogress/".
* Second, an option for the bucket to use for temp blobs, which would not be
required. If not supplied, the same bucket would be used for temp blobs as for
the associated final blobs.
I was also planning on appending a UUID string to the end of temp blob
locations, in order to guarantee their uniqueness during the temp-blob writing
phase, in case somehow multiple writes to the same final blob were in progress
at the same time.
So, with the defaults, a write to a blob {{gs://final_bucket/foo/bar}} would
use a temp blob location of, say,
{{gs://final_bucket/.inprogress/foo/bar/1157f422-32af-4e32-a797-2a0a05f28ecf}}.
The prefix could be configured via the first option; also, if the user wanted
to write all temp blobs to a "temp_bucket" bucket, that bucket could be
specified via the second option, yielding
{{gs://temp_bucket/.inprogress/foo/bar/1157f422-32af-4e32-a797-2a0a05f28ecf}}.
As for cleaning the temporary blobs, is that what
[RecoverableWriter.cleanupRecoverableState|https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/core/fs/RecoverableWriter.html#cleanupRecoverableState-org.apache.flink.core.fs.RecoverableWriter.ResumeRecoverable-]
does? I was thinking to clean up any temporary blobs there. Beyond that, if a
separate bucket for temporary blobs were used, I suppose one could apply an
[object lifecycle-management
rule|https://cloud.google.com/storage/docs/lifecycle] that would delete blobs
after some period of time. These rules look to be applicable only at the bucket
level, so this would only work if a separate bucket were used just for
temporary blobs.
*Per [this doc|https://cloud.google.com/storage/docs/resumable-uploads], a
resumable upload must be completed within a week. This could be surprising for
the users, if they try to restore a job from checkpoints/savepoints after
pausing for more than one week.*
Yes, I would propose to disclose this limitation via a disclaimer, similar to
the one used for S3:
{quote}Important Note 2: 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 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.
{quote}
Admittedly, it does seem that S3 provides more configuration options here than
GCS. It would be nice if the week limit were configurable, but it doesn't seem
to be, based on my read.
*Relying on Java serialization means depending our compatibility on the
compatibility of GCS, which should be avoid if possible. Would it be possible
to directly work with the REST API and session URI? IIUC this is how the write
channel internally works.*
I'd need to look into it more closely, but yes, I think this could be possible.
I think we'd wind up reimplementing much of what is done in
[BlobWriteChannel|https://github.com/googleapis/java-storage/blob/master/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java]
and
[BaseWriteChannel|https://github.com/googleapis/java-core/blob/master/google-cloud-core/src/main/java/com/google/cloud/BaseWriteChannel.java],
which I suppose I was thinking would be good to avoid. The code in the
BlobWriteChannel regarding retries/flushing looks a bit complicated, but
perhaps we wouldn't need all that complication in the Flink case?
In the event we were to go this route, is there a preferred client to use in
Flink for HTTP/REST requests?
> Create RecoverableWriter for GCS
> --------------------------------
>
> Key: FLINK-11838
> URL: https://issues.apache.org/jira/browse/FLINK-11838
> Project: Flink
> Issue Type: New Feature
> Components: Connectors / FileSystem
> Affects Versions: 1.8.0
> Reporter: Fokko Driesprong
> Assignee: Galen Warren
> Priority: Major
> Labels: pull-request-available, usability
> Fix For: 1.13.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> GCS supports the resumable upload which we can use to create a Recoverable
> writer similar to the S3 implementation:
> https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
> After using the Hadoop compatible interface:
> https://github.com/apache/flink/pull/7519
> We've noticed that the current implementation relies heavily on the renaming
> of the files on the commit:
> https://github.com/apache/flink/blob/master/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java#L233-L259
> This is suboptimal on an object store such as GCS. Therefore we would like to
> implement a more GCS native RecoverableWriter
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
