-
Notifications
You must be signed in to change notification settings - Fork 1.4k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add s3 support for staging local dependencies in sparkctl #144 #179
Conversation
Do you have some suggestions to the code itself? I am not sure if I can start adding documentation and update go-deps definitions, as I am not sure whether implementation needs some adjustments. |
@mrow4a GCS also supports regional access and allows users to specify a region. I plan to use the new flag |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please run go fmt
to format the code.
sparkctl/cmd/s3.go
Outdated
"github.com/aws/aws-sdk-go/aws" | ||
"github.com/aws/aws-sdk-go/aws/session" | ||
"github.com/aws/aws-sdk-go/service/s3" | ||
"path/filepath" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Golang built in packages should be on top.
sparkctl/cmd/s3.go
Outdated
if err != nil { | ||
return nil, err | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Unnecessary extra line.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
sparkctl/cmd/s3.go
Outdated
|
||
|
||
uploader := &s3Uploader{ | ||
client: s3manager.NewUploader(sess), |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does the client need to be closed when done?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I did not find any note that this is necessary in aws sdk
f401977
to
e1bbd0d
Compare
sparkctl/README.md
Outdated
``` | ||
|
||
NOTE: In Spark 2.3.0 with init-containers used for Spark with Kubernetes as resource manager, | ||
the dependencies are not mounted in proper location (e.g. application file is not mounted in jars folder) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It seems the problem I had previously with s3a deps - the init container incorrectly fetches (mounts) e.g. main application file. I think I need to investigate issue more with spark/master @liyinan926
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Spark 2.4.0 (master branch) solves the issue
sparkctl/cmd/s3.go
Outdated
return "" | ||
} | ||
|
||
return fmt.Sprintf("%s://%s.%s/%s", |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I ensured to be fully compatible with AWS, and thus it requires <endpoint-protocol>://<bucket>.<endpoint-url>/path/to/file
scheme.
@liyinan926 I have been debugging S3 issue with Spark 2.3.0 I found that the S3 dependencies are wrongly fetched because of spark-init container. I have been testing with Spark 2.4.0 (apache/spark:master) and all works fine. Please see the analysis log Configuration
LOG with Spark 2.3.0 LOCAL (Docker image "gitlab-registry.cern.ch/db/spark-service/docker-registry/spark-on-k8s:v2.3.0-xrootd-s3"):
LOG with Spark 2.3.0 KUBERNETES (Docker image "gitlab-registry.cern.ch/db/spark-service/docker-registry/spark-on-k8s:v2.3.0"):
LOG with Spark 2.4.0 KUBERNETES (Docker image "gitlab-registry.cern.ch/db/spark-service/docker-registry/spark-on-k8s:master"):
|
a923b09
to
d480f30
Compare
@mrow4a just came back from vacation. Will take a look soon. |
@liyinan926 You can try with S3 interoperability - but generally I got error in Spark 2.3 on fetching from s3, and needed to build Spark from master branch (without init containers). |
d480f30
to
afb6bab
Compare
@mrow4a This is worth taking a look at: https://github.com/google/go-cloud. It can potentially simplifies the code around uploading to GCS/S3. |
@liyinan926 will check it, so far looks promising. |
afb6bab
to
d0e5053
Compare
@liyinan926 I spent some time on implementing Should I make common code for upload, and dedicated for setting ACLs? (I guess at some point package will have ACL support) I will add new commit here to show-case it |
d0e5053
to
75b7b7e
Compare
@liyinan926 could you have a look into s3 implementation (I for now skipped GSC) |
@mrow4a I will take a look soon. |
@liyinan926 I have been playing with GCS client, and this is not trivial to set ACLs (meaning that |
75b7b7e
to
205184d
Compare
070ca2c
to
3730b25
Compare
@liyinan926 I added and tested the support for S3A and GCS using go-cloud (including documentation) - please review. |
3730b25
to
87415c6
Compare
df7a8c1
to
405b290
Compare
Add aws sdk as dependency
405b290
to
c352e1c
Compare
I did rebase and updated PR Description in the top. |
sparkctl/cmd/create.go
Outdated
@@ -31,15 +31,19 @@ import ( | |||
"k8s.io/apimachinery/pkg/util/yaml" | |||
clientset "k8s.io/client-go/kubernetes" | |||
|
|||
"context" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
These two imports should be in their respective groups.
sparkctl/cmd/create.go
Outdated
fileName := filepath.Base(localFilePath) | ||
uploadFilePath := filepath.Join(uploadPath, fileName) | ||
// Check if exists by trying to fetch metadata | ||
_, err := bh.bucket.NewRangeReader(bh.ctx, uploadFilePath, 0, 0) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We need to make sure the reader gets closed.
reader, err := bh.bucket.NewRangeReader(bh.ctx, uploadFilePath, 0, 0)
defer reader.Close()
sparkctl/cmd/create.go
Outdated
if err != nil { | ||
return "", fmt.Errorf("Failed to read file: %s", err) | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Need to add defer data.Close()
.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
data is not a handle, is actual bytes (buf.Bytes())
sparkctl/cmd/create.go
Outdated
return "", fmt.Errorf("Failed to read file: %s", err) | ||
} | ||
|
||
w, err := bh.bucket.NewWriter(bh.ctx, uploadFilePath, nil) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ditto. Need to make sure w
gets closed, by adding defer w.Close()
after the error check below.
sparkctl/cmd/create.go
Outdated
return nil, fmt.Errorf( | ||
"unable to upload local dependencies: no upload location specified via --upload-to") | ||
} | ||
|
||
uploadLocationUrl, err := url.Parse(UploadTo) | ||
var bh *BlobHandler |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks like bh
and ctx
are not needed until the start of the switch
block.
sparkctl/cmd/gcs.go
Outdated
func (blob GCSBlob) setPublicACL( | ||
ctx context.Context, | ||
bucket string, | ||
filePath string) error { | ||
client, err := storage.NewClient(ctx) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So we still need to use this gcs-specific client for setting public ACL?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes. This already kills a lot of redundant code, but ACL are not implemented in cloud-go
yet. This cloud-go
seems a very young project.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
OK, can we add a TODO
to get rid of this gcs-specific code once cloud-go
adds support for that?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
c352e1c
to
f2aa39b
Compare
this change would let spark use s3 as a hadoop endpoint(checkpoint), yes? |
sparkctl/cmd/create.go
Outdated
uploadFilePath := filepath.Join(uploadPath, fileName) | ||
// Check if exists by trying to fetch metadata | ||
reader, err := bh.bucket.NewRangeReader(bh.ctx, uploadFilePath, 0, 0) | ||
defer reader.Close() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It looks like you don't even need to use defer
here since it's closed immediately after being created.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Not sure if I understand.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You don't need to use defer
as reader
is not used anywhere below, it's closed right after it's created.
sparkctl/cmd/create.go
Outdated
// Write data to bucket and close bucket writer | ||
_, writeErr := w.Write(data) | ||
if err := w.Close(); err != nil { | ||
return "", fmt.Errorf("Failed to close bucket writer: %s", err) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We use lower case letter for the first word of the error string in this package.
sparkctl/cmd/create.go
Outdated
|
||
// Check if write has been successful | ||
if writeErr != nil { | ||
return "", fmt.Errorf("Failed to write to bucket: %s", err) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ditto. Failed
-> failed
.
@puneetloya It will allow to do the same as for GCS. Specificaly, to upload your local dependency to your AWS S3 or other S3 compatible storage using a @liyinan926 I think not to confuse people, I need to add scheme as |
That's a good idea. |
f2aa39b
to
5d4edd7
Compare
Clarified the README and adjusted to review @liyinan926 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM after the latest comments get addressed.
sparkctl/cmd/create.go
Outdated
fileName := filepath.Base(localFilePath) | ||
uploadFilePath := filepath.Join(uploadPath, fileName) | ||
// Check if exists by trying to fetch metadata | ||
_, err := uh.b.NewRangeReader(uh.ctx, uploadFilePath, 0, 0) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry if I didn't make it clear. You need to close the returned Reader
, but you don't need to use defer reader.Close()
. Simply do the following is fine.
reader, err := uh.b.NewRangeReader(uh.ctx, uploadFilePath, 0, 0)
reader.Close()
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oh, ok - now I got your point.
sparkctl/README.md
Outdated
spark.hadoop.fs.s3a.impl org.apache.hadoop.fs.s3a.S3AFileSystem | ||
``` | ||
|
||
NOTE: In Spark 2.3.0 init-containers are used for Kubernetes as resource manager. In future versions, init-containers are removed. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
for Kubernetes as resource manager
-> for downloading remote application dependencies
. Also this applies to any Spark 2.3.x versions.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
sparkctl/cmd/create.go
Outdated
@@ -267,28 +272,118 @@ func isContainerLocalFile(file string) (bool, error) { | |||
return false, nil | |||
} | |||
|
|||
type Blob interface { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks like both Blob
and UploadHandler
are only used inside the cmd
package, so they should both start with lower-case letters.
Add documentation for staging local dependencies in s3a
5d4edd7
to
23f295d
Compare
Description
Adds support for staging local dependencies in sparkctl (solves #144 )
--upload-to-endpoint
set tohttps://storage.googleapis.com
and--upload-to-region
default empty regionTo bo done
How did I test
@liyinan926 @prasanthkothuri