I’ve submitted a Flink Improvement Proposal (FLIP) about adding support for Flink to be able to fetch remote jars for applications submitted to a session cluster.
Discussion thread on lists.apache.org
Background
I think it’ll help if I start by describing the sort of scenario my FLIP is trying to help with. This is how Flink works today…
(If you already know about Flink’s artifact fetcher, you should probably skip this bit.)
Fetching jars for Application mode
Let’s say that Alice’s team write Flink applications, and the jars for their apps are stored in their team’s S3 bucket.
They can run their Flink Applications in Kubernetes by creating specs like this:
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: aliceteam-app-1
spec:
image: flink-with-s3-fs-plugin:1.0
flinkConfiguration:
s3.endpoint: http://enterprise.artifactstore:9000
s3.path.style.access: "true"
s3.access-key: aliceteam-access
s3.secret-key: aliceteam-secret
job:
entryClass: com.example.datastream.aliceteam.SalesJob
jarURI: s3://aliceteam-artifacts/example-1-app.jar
...
Flink has an artifact fetcher class that can download a jar from remote storage. The Job Manager pod comes up, gets the jar from the remote storage, and starts running it.
As a diagram, that might look a bit like this:
This is oversimplified, because obviously the Operator does do a lot to manage and maintain the Flink cluster. What I’m trying to show here is that for the purposes of fetching the jar, the Operator doesn’t need to do anything. Bob (the admin who runs the Flink Kubernetes Operator in the company Kubernetes cluster) doesn’t need to do anything.
If Charlie’s team wants to run their Flink apps, and their jars are also stored in S3, they can do something like this. Their own Flink deployment can use their own credentials to access their jar.
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: charlieteam-app-1
spec:
image: flink-with-s3-fs-plugin:1.0
flinkConfiguration:
s3.endpoint: http://enterprise.artifactstore:9000
s3.path.style.access: "true"
s3.access-key: charlieteam-access
s3.secret-key: charlieteam-secret
job:
entryClass: com.example.datastream.charlieteam.LogisticsJob
jarURI: s3://charlieteam-artifacts/example-1-app.jar
...
The result would look a bit like this:
Still nothing for admin Bob to do.
If Jane’s team wants to run their Flink apps, but they store jars in Azure blob storage, they could do something like this:
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: janeteam-app-1
spec:
image: flink-with-blob-fs-plugin:1.0
flinkConfiguration:
fs.azure.account.key.janeteamartifacts.blob.core.windows.net: janeteam-secret-key
job:
entryClass: com.example.datastream.jane.AnalyticsJob
jarURI: wasbs://[email protected]/example-1-app.jar
...
Their Flink image would be built with an Azure storage plugin instead. And adding their deployment would look a bit like this:
And still, no issues for admin Bob. Each team can be self-sufficient in deploying their applications.
This assumes they all run their Flink applications in per-job application mode clusters.
But what if each team wants to run their apps in per-team session mode clusters?
Fetching jars for Session mode
Flink’s artifact fetcher isn’t wired into job submission, it’s only accessible from cluster startup paths. But a session cluster is already running, so that doesn’t help. When submitting an application to a Flink cluster in session mode, the jar needs to be made available locally.
The Flink Kubernetes Operator I’m using in my examples here has what is sort of a workaround for this. Alice’s teams can define their applications like this:
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
spec:
deploymentName: alice-team-cluster
job:
entryClass: com.example.datastream.aliceteam.SalesJob
jarURI: s3://aliceteam-artifacts/example-1-app.jar
...
If she does this, before submitting the Job to the alice-team-cluster, the Operator will download the jar from the shared S3 storage to the Operator pod. From there, it will upload it to the Job Manager for the session cluster, and only then submit the Job to Flink, with the jar now locally available as needed.
That would look like this:
For this to work, Alice’s team will need to ask Bob, the admin running the Operator, to add an S3 filesystem plugin to the Operator, and to configure it with credentials that will let it download their jars.
If Charlie’s team also want to start running their apps in their own team session cluster, they’ll do something like this:
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
spec:
deploymentName: charlie-team-cluster
job:
entryClass: com.example.datastream.charlieteam.LogisticsJob
jarURI: s3://charlieteam-artifacts/example-1-app.jar
...
That would look like this:
This means that admin Bob needs to put S3 credentials in the Operator that has read access to the S3 buckets for both Alice’s team and Charlie’s team. The S3 credentials they put in the Operator needs access to everything it might be asked to download from S3.
When Jane’s team want to switch to using session clusters, they can do something similar.
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
spec:
deploymentName: jane-team-cluster
job:
entryClass: com.example.datastream.jane.AnalyticsJob
jarURI: wasbs://[email protected]/example-1-app.jar
...
As with the others, the Operator will need to download the jar first and then upload it to their session cluster.
For the Operator to be able to download jars for Jane’s team, they’ll need to ask admin Bob to update the Operator again – to add a blob storage filesystem plugin, and configure it with credentials it can use.
What we could improve
This is a useful feature, but there are some challenges with this approach.
Firstly, there are some rough edges. For example, there is likely a bit of a delay introduced by the Operator having to download the jar, and then upload it to the Job Manager pod (rather than the Job Manager just download it for itself). You end up with two copies of the jar in two places.
These are, at most, rough edges and probably not deal breakers.
The bigger challenge is the need to store credentials with access to everything in the Operator. In environments where the Operator is managing Flink clusters in different environments, for different logical tenants and different teams, the need to have such widely scoped credentials in one place is a potential security issue.
It also means each team running Flink session clusters can’t be independently self-sufficient, as they are dependent on the admin running the Operator to update it to be able to download their jars.
The proposal
The ideal would be for session clusters to support the same behaviour as it does for per-job application mode clusters. Flink already has an artifact fetcher that is able to retrieve a jar for a new job, so if we wire that existing logic into job submission on session clusters, we’ll get the best of both worlds.
Alice’s team would create their own session cluster, and configure it with the filesystem plugin and credentials it needs to access their team applications.
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: alice-team-cluster
spec:
image: flink-with-s3-fs-plugin:1.0
flinkConfiguration:
web.submit.jar-uri.enable: "true"
s3.endpoint: http://enterprise.artifactstore:9000
s3.path.style.access: "true"
s3.access-key: aliceteam-access
s3.secret-key: aliceteam-secret
...
Each of their applications could be created like this:
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
name: aliceteam-app-1
spec:
deploymentName: alice-team-cluster
job:
entryClass: com.example.datastream.aliceteam.SalesJob
flinkConfiguration:
user.artifacts.job-jar: s3://aliceteam-artifacts/example-1-app.jar
...
(Notice that to avoid the Flink Kubernetes Operator helpfully applying it’s workaround of trying to download the jar, I’m putting the job jar URI in a flink config value, so it’s passed to the Job Manager for it to do the download instead.)
Each team can run their own session cluster, where they have the freedom to specify the credentials to use and the filesystem plugins for where they store artifacts.
And the Operator can agnostically and independently manage all of them, without needing any elevated credentials of its own.
The FLIP
I’ve written up the spec of how I’m proposing to implement all of this, which goes into more detail about the config options I think we’ll need and the pros and cons of adding it.
I hope it’ll be a useful enhancement for users in the sort of scenario I’ve described above. If you run your own Kubernetes Operator to manage your own Flink session clusters, then this probably doesn’t make much of a difference to you. It means you could put your filesystem credentials in a different place, and you might notice a tiny difference in deployments by avoiding needing to download-then-upload your app jars. But that’s about it.
Although the scenarios I’m describing here focus on using the Flink Kubernetes Operator, the changes I’m suggesting to Flink are mostly in flink-runtime-web.
The Operator will be able to take advantage of this without needing any changes, but so will anything else that submits jobs to session clusters and currently has to upload the jar to the Job Manager first.
What’s next?
It is time to see if the Flink community feel like this would be useful.
I’ve written up a more detailed proposal in a Google Doc and asked for feedback on the Flink dev mailing list.
If you have opinions on any of this, please do join the discussion – or you can let me know directly.
Tags: apacheflink, flink, flip