beam-flink-k8s

Example project for running Beam on Flink in Kubernetes based on Müller Fourier's Medium article.

Requirements

• kubectl installed

Local Deployment

Additional Requirements

• docker installed
• minikube installed

Preparations

Docker service must be running:

start minikube

eval $(minikube -p minikube docker-env)

kubectl get all

EKS Deployment

Additional Requirements

• eksctl installed

Preparations

Activate AWS profile if needed:

export AWS_PROFILE=<your_profile>

Set AWS region:

export AWS_REGION=<your_region>

Set KMS key used for envelope encryption of Kubernetes secrets: (create a new customer managed KMS key if needed)

export AWS_KMS_KEY_EKS=<your_key_arn>

Launch cluster (envsubst will substitute the evironment variables in the .yaml file for you):

envsubst < flink_cluster/eks-flink-cluster.yaml | eksctl create cluster -f -

Check that your nodes have been created:

kubectl get nodes

Update the kubeconfig file to interact with you cluster:

aws eks update-kubeconfig --name beam-flink-eks

Spin up Flink cluster

Create configuration and servive definitions:

kubectl apply -f flink_cluster/flink-configuration-configmap.yaml

kubectl apply -f flink_cluster/jobmanager-service.yaml

Launch Jobmanager (orchestrator) and Taskmanager (worker) deployments:

kubectl apply -f flink_cluster/jobmanager-session-deployment.yaml

kubectl apply -f flink_cluster/taskmanager-session-deployment.yaml

Run Beam job

Check that jobmanager and taskmanager deployments & pods are ready:

kubectl get all

Init env variable to point to the endpoint of the jobmanager

export JOBMANAGER_ENDPOINT=$(k get pods -l app=flink,component=jobmanager -o jsonpath='{.items[].status.podIP}')

Submit job (envsubst will substitute the JOBMANGER_ENDPOINT for you):

envsubst < beam_jobs/word_count/beam_wordcount_py.yaml | kubectl apply -f -

Monitor jobs by port forwarding and opening localhost:8081 for Flink dashboard

kubectl port-forward <service/flink-jobmanager> 8081:8081

Delete job (if needed):

kubectl delete -f beam_jobs/word_count/beam_wordcount_py.yaml

Shutdown Cluster

kubectl delete -f flink_cluster/taskmanager-session-deployment.yaml
kubectl delete -f flink_cluster/jobmanager-session-deployment.yaml
kubectl delete -f flink_cluster/jobmanager-service.yaml
kubectl delete -f flink_cluster/flink-configuration-configmap.yaml

Local Deployment

minikube stop

EKS Deployment

eksctl delete cluster --name beam-flink-eks