# Flink is a distributed runner, hoping to use it with beam

What a mess the documentation is.

## RBAC

Trying to avoid using the default service account, so running:

```
kubectl create serviceaccount flink-service-account
kubectl create clusterrolebinding flink-role-binding-flink --clusterrole=edit --serviceaccount=default:flink-service-account

# Save for later
kubectl get serviceaccount flink-service-account -o yaml > 1.account.yaml
kubectl get clusterrolebinding flink-role-binding-flink -o yaml > 2.role-binding.yaml
```

Unknown if those will yaml files will work to recreate, but worth a try.

Set the service account in the configuration

```
cd flink-1.17.1
echo '#====================================' >> conf/flink-conf.yaml
echo '# Freds Custom Configuration' >> conf/flink-conf.yaml
echo '#====================================' >> conf/flink-conf.yaml
echo '' >> conf/flink-conf.yaml
echo 'kubernetes.jobmanager.service-account: flink-service-account' >> conf/flink-conf.yaml
```

## Java

```
sudo apt install default-jre-headless
```

## Deployment

https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/resource-providers/native_kubernetes/


```
# I needed to pull a different image (error: "exec format error").
echo 'kubernetes.container.image.ref: arm64v8/flink:1.17.1-scala_2.12' >> conf/flink-conf.yaml
echo 'kubernetes.rest-service.exposed.type: LoadBalancer' >> conf/flink-conf.yaml
```


```
# Create the cluster
./bin/kubernetes-session.sh -Dkubernetes.cluster-id=my-first-flink-cluster
# On 1.14, the image override didn't work, so instead:
./bin/kubernetes-session.sh \
    -Dkubernetes.cluster-id=my-first-flink-cluster \
    -Dkubernetes.container.image.ref=arm64v8/flink:1.13.6-scala_2.12

# If you want to Delete what you just created
kubectl delete deployment/my-first-flink-cluster

# Run a test job:
./bin/flink run \
    --target kubernetes-session \
    -Dkubernetes.cluster-id=my-first-flink-cluster \
    ./examples/streaming/TopSpeedWindowing.jar

# To reconnect (wtf?)
./bin/kubernetes-session.sh \
    -Dkubernetes.cluster-id=my-first-flink-cluster \
    -Dexecution.attached=true
# Type "stop" to stop it.
```

# Let's beam to it

```
cd .. # Back out of the flink dir

sudo apt install python3.10-venv
python -m venv env
source env/bin/activate
pip install -r requirements.txt
```

Issue: Might need an older flink version
Latest compatible according to: https://beam.apache.org/documentation/runners/flink/ appears to be 1.14.x

# Yup. Need older version

Note: Image ref is different in older version.

from `conf/flink-conf.yaml`:

```
#====================================
# Freds Custom Configuration
#====================================

kubernetes.jobmanager.service-account: flink-service-account
kubernetes.container.image: arm64v8/flink:1.14.6-scala_2.12
kubernetes.rest-service.exposed.type: LoadBalancer
```

## Beam test, take 2

```
# Start the flink server:
cd ~/src/monkeybox_kubernetes/Workloads/flink
source env/bin/activate
flink-1.14.6/bin/kubernetes-session.sh -Dkubernetes.cluster-id=my-first-flink-cluster
# Ignore the Job Manager Web Interface - It will be moved to a VIP
python3 beam_test.py
```

Still didn't work..  Lots of issues. It's like beam really doesn't want to play this way.
Renaming this readme and trying a different way.