설치 순서는 ES -> Kibana -> FluentBit 순이다. Fluentd 를 먼저 시도했는데 ES 로 데이터가 유입되지 않았다. 유저 불량으로 보이긴 하는데 아무런 오류 메시지도 보여주지 않아서 원인 확인이 어려웠다. FluentBit 은 매끄럽게 바로 동작했다.

ES 와 Kibana 는 Helm chart 로 설치한다.

먼저 ES.

helm repo add elastic https://helm.elastic.co
kubectl create namespace elastic
helm install elasticsearch elastic/elasticsearch -n elastic

잠깐 테스트하고 뺄 예정이었어서 기본값으로 설치했는데, Minikube 에 작은 사이즈로 설치하려면 아래 명령으로 helm install 을 해주자.

curl -O https://raw.githubusercontent.com/elastic/Helm-charts/master/elasticsearch/examples/minikube/values.yaml
helm install --name elasticsearch elastic/elasticsearch -f ./values.yaml

설치가 완료되었으면

kubectl get all -n elastic

kubectl get pods --namespace=elastic -l app=elasticsearch-master -w

으로 설치된 object 들을 확인하자. 서비스에 elasticsearch-master 가 있어야 한다. 그리고 모든 pods 가 running 으로 전환되는 것을 확인하자.

서비스에 대해 port-forward 를 하고, index 리스트를 조회해서 정상적으로 동작하는지 확인하자.

kubectl port-forward -n elastic svc/elasticsearch-master 9200

curl http://localhost:9200/_cat/indices

정상적으로 동작한다면 Kibana 설치로 넘어가자.

helm install kibana elastic/kibana -n elastic

kibana-kibana 가 deployment 에 등록되어 있을 것이다. 같은 이름으로 서비스도 등록되어 있다. port-forwarding 하고 브라우저에서 열어서 Kibana 페이지가 정상적으로 열리는지 확인하자.

kubectl port-forward deployment/kibana-kibana 5601

이제 ES 와 Kibana 는 설치되었다. FluentBit 을 설치해 보자.

FluentBit 은 helm chart 를 사용하지 않고 daemonset 을 직접 등록한다.

1) service account (service-account.yaml)

apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluent-bit
  namespace: elastic

2) role (role.yaml)

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluent-bit-read
rules:
- apiGroups: [""]
  resources:
  - namespaces
  - pods
  verbs: ["get", "list", "watch"]

3) cluster role binding (role-binding.yaml)

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluent-bit-read
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: fluent-bit-read
subjects:
- kind: ServiceAccount
  name: fluent-bit
  namespace: elastic

4) configmap (configmap-1.yaml)

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluent-bit-config
  namespace: elastic
  labels:
    k8s-app: fluent-bit
data:
  fluent-bit.conf: |
    [SERVICE]
        Flush         1
        Log_Level     info
        Daemon        off
        Parsers_File  parsers.conf

@INCLUDE input-kubernetes.conf
    @INCLUDE output-elasticsearch.conf
  input-kubernetes.conf: |
    [INPUT]
        Name              tail
        Tag               kube.*
        Path              /var/log/containers/*.log
        Parser            docker
        DB                /var/log/flb_kube.db
        Mem_Buf_Limit     5MB
        Skip_Long_Lines   On
        Refresh_Interval  10

output-elasticsearch.conf: |
    [OUTPUT]
        Name            es
        Match           *
        Host            ${FLUENT_ELASTICSEARCH_HOST}
        Port            ${FLUENT_ELASTICSEARCH_PORT}
        Logstash_Format On
        Logstash_Prefix fluent-bit
        Retry_Limit     False

parsers.conf: |
    [PARSER]
        Name        docker
        Format      json
        Time_Key    time
        Time_Format %Y-%m-%dT%H:%M:%S.%L
        Time_Keep   On

읽을 파일 경로, 파서 규칙, 기록할 위치 등을 변경할 때 여길 변경하면 되겠다. 기본 설정은 모든 container 가 기록하는 로그 파일을 읽어들이도록 되어 있다. (/var/log/containers 아래 container 의 stdout/stderr 출력이 파일로 기록됨.)

5) daemonset (daemonset.yaml)

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluent-bit
  namespace: elastic
spec:
  selector:
    matchLabels:
      k8s-app: fluent-bit-logging
  template:
    metadata:
      labels:
        k8s-app: fluent-bit-logging
    spec:
      containers:
      - name: fluent-bit
        image: fluent/fluent-bit:1.3.8
        imagePullPolicy: Always
        env:
        - name: FLUENT_ELASTICSEARCH_HOST
          value: "elasticsearch-master"
        - name: FLUENT_ELASTICSEARCH_PORT
          value: "9200"
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
        - name: journal
          mountPath: /journal
          readOnly: true
        - name: fluent-bit-config
          mountPath: /fluent-bit/etc/
      terminationGracePeriodSeconds: 10
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: journal
        hostPath:
          path: /var/log/journal
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
      - name: fluent-bit-config
        configMap:
          name: fluent-bit-config
      serviceAccountName: fluent-bit
      tolerations:
      - key: node-role.kubernetes.io/master
        effect: NoSchedule

fluent-bit-config 라는 이름으로 위의 config map 을 참조하고 있음을 알 수 있다.

순서대로 반영하자.

kubectl apply -f service-account.yaml
kubectl apply -f role.yaml
kubectl apply -f role-binding.yaml
kubectl apply -f configmap-1.yaml
kubectl apply -f daemonset.yaml

정상적으로 반영되었다면 각 node 별로 fluent-bit-* 이라는 pod 이 실행되고 있을 것이다. running 상태인지 한 번 확인해 주고, ES 의 index 를 다시 조회해 보자.

curl http://localhost:9200/_cat/indices

결과 라인 중 green open fluent-bit-날짜 … 가 있다면 성공이다.

이제 Kibana 페이지를 열어서, Management -> Stack Management -> Kibana -> Index Patterns 를 선택한다.

페이지에서 우측 상단의 Create index patterns 버튼을 누르고, 다음 페이지에서 index pattern name 에 “fluent-bit-*” 을 입력하면 아래 pattern 에 맞는 인덱스 리스트가 나타난다. (현재는 하나일 테고 날짜에 따라 계속 쌓이게 되겠다.) 원하는 결과가 나왔다면 Next Step 버튼을 눌러준다.

다음 페이지에서 Time field 에 @timestamp 를 선택하고, Create index pattern 버튼을 눌러 완성한다.

Kibana -> Discover 페이지로 돌아가면, 왼쪽 상단 index pattern 에 fluent-bit-* 로 나오고 필드 리스트가 나타날 것이다. 오른쪽에는 유입이 진행되고 있다면 메시지들이 업데이트 되고 있을 것이다. 혹시 메시지가 업데이트 되지 않는다면 Spark 어플리케이션 하나 제출해서 실행하면 업데이트 될 것이다.

References

• Deploying the ELK Stack on Kubernetes with Helm
• How to Set up Log Forwarding in a Kubernetes Cluster Using Fluent Bit

일단 k8s 클러스터에 Prometheus 설치부터 하자.

kubectl create namespace prometheus

helm install prometheus stable/prometheus-operator -n prometheus

설치가 되었다면, k8s 에 Prometheus 관련 여러 개의 서비스들이 등록되어 있을 것이다.

kubectl get svc -n prometheus

NAME                                      TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                      AGE
prometheus-prometheus-oper-alertmanager   ClusterIP   10.43.133.105   <none>        9093/TCP                     71s
prometheus-prometheus-oper-prometheus     ClusterIP   10.43.50.232    <none>        9090/TCP                     71s
prometheus-kube-state-metrics             ClusterIP   10.43.23.65     <none>        8080/TCP                     71s
prometheus-prometheus-oper-operator       ClusterIP   10.43.107.212   <none>        8080/TCP,443/TCP             71s
prometheus-prometheus-node-exporter       ClusterIP   10.43.66.204    <none>        9100/TCP                     71s
prometheus-grafana                        ClusterIP   10.43.254.135   <none>        80/TCP                       71s
alertmanager-operated                     ClusterIP   None            <none>        9093/TCP,9094/TCP,9094/UDP   60s
prometheus-operated                       ClusterIP   None            <none>        9090/TCP                     50s

“prometheus-prometheus-oper-prometheus” 이 서비스를 port-forwarding 하고, 브라우저를 통해 열어 보면 Prometheus 가 정상적으로 실행 중임을 확인할 수 있다.

kubectl port-forward -n prometheus service/prometheus-prometheus-oper-prometheus 9090

Grafana 도 한 세트로 설치가 된다. “prometheus-grafana” 를 port-forwarding 하고 브라우저를 통해 열어 보자.

kubectl port-forward -n prometheus service/prometheus-grafana 9091:80

기본적으로 id 는 admin, password는 prom-operator 로 설정되어 있다. 어떤 chart/operator 를 설치하냐에 따라 password 가 제각각이다. 구글링으로 한 번 더 확인하자.

기본적으로 설치되어 있는 Prometheus 가 data source 로 등록되어 있다. k8s 클러스터를 모니터링 할 수 있는 dashboard 들도 등록되어 있으니 Dashboard -> Manage 에서 둘러보자.

Prometheus 는 설치가 되었으니 Spark 에서 metrics 를 넣어 보자.

Prometheus 는 기본적으로 push 가 아닌 pull 방식으로 동작한다. 다른 말로 하면, Spark 에서 metric 정보를 노출해 주고, prometheus 에서 해당 노출 경로로 접근해서 데이터를 가져가게 된다. (push gateway 를 설치하고 여기로 push 하는 방안도 있다고 한다. 공식 문서에 따르면 일부 유즈 케이스를 제외하면 추천하지 않는 듯)

Spark 의 Prometheus 공식 지원은 두 가지로 이루어진다.

1. Prometheus Metric Sink 추가
2. UI 를 통한 Prometheus 전용 executor metrics 노출 켜기

1번은 Spark 이 Dropwizard 를 활용하여 노출하는 metric 을 Prometheus 가 가져갈 수 있는 형태로 노출하는 것이다. pull 방식이므로 여타 time-series DB 에 연동하는 방법과는 다르게 전용 Servlet 을 생성하고 UI 에 붙여 준다.

설정 방법은 다음과 같다.

1. metrics.properties 파일 수정

템플릿 파일에도 설정이 주석처리 되어 있을 것이다. 어려울 게 없다.

*.sink.prometheusServlet.class=org.apache.spark.metrics.sink.PrometheusServlet
*.sink.prometheusServlet.path=/metrics/prometheus

2. 해당 파일을 Spark docker image 에 포함 vs 어플리케이션 제출 시에 파일 포함시키고 설정에서 해당 파일을 metric 설정 파일로 사용하도록 변경하기

모든 어플리케이션을 Prometheus 로 연동할 예정이고, 무엇보다 쉽게 가고 싶으면 전자로 가자. 설명하겠지만 (그리고 실제로 이렇게 설정해 두었지만) 후자가 좀 귀찮게 되어 있다. 대신 설정 파일 내용 변경하고 이미지만 build/push 하는 거니까 설명은 생략.

(SO 에 따르면) 후자의 가장 쉬운 방법은 --files 에 설정이 완료된 metrics.properties 파일을 추가하고 --conf spark.metrics.conf=./metrics.properties 로 추가된 파일을 사용하도록 설정하는 것이다. Scala 로 해보진 않았지만, PySpark 3.1.0 SNAPSHOT, k8s cluster mode 기준으로 이 방법은 동작하지 않았다. SparkContext 를 초기화할 때 메트릭 설정이 먼저 발생하고 (파일 없음 오류 발생) 나중에 files 에 있는 파일을 로드하는 것으로 보인다.

여러 방안들이 더 있겠지만, 기 설정해 둔 NFS 를 여기에 활용할 수 있다. 디렉토리를 하나 더 만들어서 (spark-resources 라고 하자) NFS 설정을 해 주고, static provisioning 방법으로 PV/PVC 등록을 해 주자.

> PV

vim nfs-pv-spark-resources.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: nfs-pv-spark-resources
  labels:
    name: nfs-pv-spark-resources
spec:
  capacity:
    storage: 1Gi
  volumeMode: Filesystem
  accessModes:
    - ReadOnlyMany
  persistentVolumeReclaimPolicy: Recycle
  storageClassName: nfs
  mountOptions:
    - hard
    - nfsvers=4.1
  nfs:
    path: /nfs/spark-resources
    server: 192.168.1.60

kubectl apply -f nfs-pv-spark-resources.yaml

> PVC

vim nfs-pvc-spark-resources.yaml

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: nfs-pvc-spark-resources
spec:
  storageClassName: nfs
  accessModes:
    - ReadOnlyMany
  resources:
    requests:
      storage: 1Gi
  selector:
    matchLabels:
      name: nfs-pv-spark-resources

kubectl apply -f nfs-pvc-spark-resources.yaml

NFS 에 설정한 /nfs/spark-resources 디렉토리에 설정이 완료된 metrics.properties 파일을 넣고, 어플리케이션 제출에 아래 설정을 추가하자.

--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.res.options.claimName="nfs-pvc-spark-resources" \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.res.mount.path="/home/spark/resources" \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.res.options.claimName="nfs-pvc-spark-resources" \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.res.mount.path="/home/spark/resources" \
--conf spark.metrics.conf=/home/spark/resources/metrics.properties \

이제 어플리케이션을 제출하고, Spark UI 를 port forwarding 하는 등의 방법으로 접근해서 /metrics/prometheus 경로를 열어 보자. 페이지가 열리면서 아래와 비슷한 metric 정보들이 출력되면 설정이 잘 된 것이다.

metrics_spark_bf62f5b6f14d4170b4ef7b58673b5dd5_driver_BlockManager_disk_diskSpaceUsed_MB_Number{type="gauges"} 0
metrics_spark_bf62f5b6f14d4170b4ef7b58673b5dd5_driver_BlockManager_disk_diskSpaceUsed_MB_Value{type="gauges"} 0
metrics_spark_bf62f5b6f14d4170b4ef7b58673b5dd5_driver_BlockManager_memory_maxMem_MB_Number{type="gauges"} 827
metrics_spark_bf62f5b6f14d4170b4ef7b58673b5dd5_driver_BlockManager_memory_maxMem_MB_Value{type="gauges"} 827
metrics_spark_bf62f5b6f14d4170b4ef7b58673b5dd5_driver_BlockManager_memory_maxOffHeapMem_MB_Number{type="gauges"} 0
...

이제 이 정보를 Prometheus 가 가져가도록 설정해야 한다.

기본 설치 시에 이미 pod monitor 와 service monitor 를 모니터링하도록 설정되어 있다.

kubectl get prometheus -A

NAMESPACE    NAME                                    VERSION   REPLICAS   AGE
prometheus   prometheus-prometheus-oper-prometheus   v2.18.2   1          4h35m

kubectl describe prometheus -n prometheus prometheus-prometheus-oper-prometheus

...

Pod Monitor Namespace Selector:
  Pod Monitor Selector:
    Match Labels:
      Release:   prometheus

...

Service Monitor Namespace Selector:
  Service Monitor Selector:
    Match Labels:
      Release:  prometheus

pod 을 감시하거나 service 를 감시하도록 monitor 를 등록해 준다. Spark 이 생성하는 service 들은 label 이 없으므로 (3.1.0 기준) service monitor 를 사용하는 경우 label 을 수동으로 등록하거나 별도 수정을 통해 자동으로 label 을 등록하게 해야 한다. 반대로, Spark 이 생성하는 pod 들은 몇 개의 label 을 자동 등록하므로 pod monitor 를 사용하면 별도 labeling 없이 pod monitor 에서 인식시킬 수 있다.

> service monitor

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  labels:
    release: prometheus
  name: spark-servicemonitor
spec:
  endpoints:
  - interval: 5s
    port: spark-ui
    path: /metrics/prometheus/
  - interval: 5s
    port: spark-ui
    path: /metrics/executors/prometheus/
  namespaceSelector:
    any: true
  selector:
    matchLabels:
      spark-role: driver

spark-ui 에 해당하는 service 에 “spark-role=driver” 라는 label 을 등록해 줘야 인식한다. 제한적으로 metric 을 유입시키고 싶을 땐 이 쪽도 방법이 되겠다.

> pod monitor

apiVersion: monitoring.coreos.com/v1
kind: PodMonitor
metadata:
  labels:
    release: prometheus
  name: spark-podmonitor
spec:
  podMetricsEndpoints:
  - interval: 5s
    port: spark-ui
    path: /metrics/prometheus/
  - interval: 5s
    port: spark-ui
    path: /metrics/executors/prometheus/
  namespaceSelector:
    any: true
  selector:
    matchLabels:
      spark-role: driver

cluster mode 로 실행한다면 driver pod 은 “spark-role=driver” 을 label 로 갖고 있다. port 도 입력해 줘야 되는데, service 에서 사용하는 alias 를 사용할 수 있는 것으로 보인다.

위에서 path 가 두 가지인데 “/metrics/executors/prometheus” 는 UI 에서 prometheus 전용 executor metric 을 노출할 때 기본 path 로 쓰는 것이다. 사용하지 않으면 제외해도 된다. (하지만 현실적으로 두 path 모두를 사용하고 싶을 것이다. 이유는 아래에 다시 언급)

이 방법의 장점은 기존 노출되던 metric 을 100% 활용할 수 있다는 것이다. (당연한 얘기로 들리겠지만 2번이 다른 접근 방법을 취하기 때문에 2번 대비 충분한 장점이다.) Structured Streaming 도 많진 않지만 별도 metric 을 옵션을 통해 노출하고 있어서 오래 실행될 쿼리를 모니터링하기엔 이 쪽이 좋다.

이 방법의 단점은 크게 두 가지이다.

1. Prometheus Sink (Servlet) 이 UI 에 의존성이 있다.

다른 말로 하면, UI 를 옵션으로 끄면 Prometheus Sink 도 노출되지 않을 것이다. 그리고 standalone 이 아닌 경우에는 driver 의 Spark UI 외에는 executor 에 UI 가 실행될 게 없다. 무슨 얘기냐면 executor 들의 metric 을 가져올 방법이 없다는 이야기 이다. push gateway 를 활용하거나, executor 도 UI 를 실행하도록 Spark 커뮤니티에 요청하는 수밖에…

2. metric key 가 엄청나게 길다.

이건 tag 를 지원하지 않는 time-series DB 들을 지원하기 위해 key 에 모든 식별 가능한 정보를 붙이다 보니 어쩔 수 없는 현상인데, tag 를 지원하는 Prometheus 입장에서는 엄청난 마이너스라 할 수 있겠다. 예를 들어 Grafana 등에서 key 로 조회할 때 key 뒷 부분이 다 잘려서 자동완성이 의미가 없어진다.

metrics_spark_bf62f5b6f14d4170b4ef7b58673b5dd5_driver_spark_streaming_7e5879b7_76d5_444c_8a63_f8fdc7491c07_inputRate_total_Number

이게 하나의 metric key 다. Structured Streaming 의 metric 은 app 식별자에 query 식별자까지 붙어서 엄청나게 길다. Prometheus 에서 relabeling 을 통해 조정이 가능한 걸로 보이는데 상세하게 보진 않았다. regex 로 하는 것 같은데 손이 엄청 갈 것 같은…

이제 UI 에서 prometheus 전용 executor metric 을 노출하는 방안을 살펴보자.

--conf spark.ui.prometheus.enabled=true

끝이다. 위에서 언급했듯이 기본 경로는 “/metrics/executors/prometheus” 이다. Spark UI 페이지를 열고 위의 경로를 붙여서 열어보자. “/metrics/prometheus” 의 페이지 내용과 비교하면 metric key 가 상당히 간결하다는 것을 알 수 있다. (tag 로 빼 두었기 때문)

기본적으로는 위의 pod monitor/service monitor 를 설정하고 “/metrics/executor/prometheus” path 를 포함시켜야 동작한다. 이미 첫 번째 방법을 소개할 때 두 번째 방법까지 처리해 둔 것이다.

여기에 더해, Spark 은 annotation 을 붙여서 Prometheus 가 알아서 가져가도록 설정할 수도 있다. (pod monitor/service monitor 를 사용하지 않는 방법)

--conf spark.kubernetes.driver.annotation.prometheus.io/scrape=true  \
--conf spark.kubernetes.driver.annotation.prometheus.io/path=/metrics/executors/prometheus/  \
--conf spark.kubernetes.driver.annotation.prometheus.io/port=4040 \

그럼 pod monitor/service monitor 는 왜 필요한가? 설정을 보면 port 한 개 path 한 개로 제한적임을 알 수 있다. 이 방법이 tag 도 지원하고 하니 이것만 쓰면 되는 거 아닌가? 왜 path 한 개만 지원하는 게 제한인가? 이 방법은 Spark REST API 에서 넘겨 주는 executor 정보에 있는 metric 만을 커버하며, Dropwizard 로 등록되는 metric 을 전혀 커버하지 않는다. 최대한 많은 metric 을 수집하고 싶다면 결국 둘 다 써야 된다. (정확하게는, 둘 다 써도 executor 측의 일부 metric 은 현재로써는 수집할 수 없다. 정말 필요하다면 push gateway 를 써야 될 듯?)

Update manifests to latest version of Prometheus Operator by brancz · Pull Request #16 · prometheus-operator/kube-prometheus

위의 PR 댓글을 따라가면서 읽어보면 Prometheus operator 에는 기능이 의도적으로 빠져 있다. 댓글 중에 values.yaml 을 직접 수정해서 annotation 을 동작하게 만드는 방안이 달려 있긴 하다.

이렇게 등록하고 나서 Prometheus UI -> Status -> Service Discovery 를 확인해서 pod monitor 에 path 별로 discovery 가 각각 존재하는지, pod 리스트에 있는 Spark driver pod 들이 UP 상태인지 (동작을 멈춘 driver pod 이라면 DOWN 이 정상일 것이다) 확인하면 끝이다.

References

• Helm chart를 활용한 Prometheus & Grafana Kubernetes에 구성하기
• Spark 3.0 Monitoring with Prometheus
• Monitoring Apache Spark on Kubernetes with Prometheus and Grafana
• WHEN TO USE THE PUSHGATEWAY

그 중 하나가 볼륨이다. 개별 워커 노드에 공간을 충분히 할애하고 emptyDir 이나 hostPath 등을 사용할 수도 있겠지만 노드에 미리 공간할애를 해 둬야 된다는 부분이 귀찮고 비효율적이다. 호스트 서버에는 할애한 공간 외에도 몇 백 기가정도의 공간이 따로 남아 있다. 어떻게 하면 좋을까?

조금 찾아보니 NFS 를 설정하고 Kubernetes 에서 PV/PVC 를 통해 해당 NFS 를 가져다 쓰는 게 서비스 비종속적인 방법 중에서는 가장 쉬워 보였다.

역시 디테일 없이 구글링해서 동작하는 걸 확인한 것만 정리했다. 디테일은 아래 문서들에서 찾자.

• How to Install NFS Client and Server on Ubuntu 20.04
• Using NFS to Share Ubuntu 20.04 Files with Remote Systems
• How to Set Up an NFS Mount on Ubuntu (Step-by-Step Guide) | DigitalOcean

우선 호스트 서버에 NFS 를 설치해야 한다. Ubuntu 20.04 기준이다.

sudo apt-get install nfs-common nfs-kernel-server rpcbind portmap
systemctl enable nfs-kernel-server
systemctl start nfs-kernel-server

다음으로 디렉토리 하나를 NFS 에 내주자.

sudo mkdir -p /nfs/vol1
sudo chown nobody:nogroup /nfs/vol1
sudo vim /etc/exports

/nfs/vol1 10.38.12.0/24(rw,sync,no_subtree_check,no_root_squash)

multipass 가 노드들의 IP 를 10.38.12.x 대역으로 내주고 있어서 그 대역에만 오픈해 주었다.

아래의 명령으로 반영하자.

sudo exportfs -r

아래의 명령으로 확인하자.

showmount -e

Export list for jungtaek-ubuntu-desktop:
/nfs/vol1 10.38.12.0/24

sudo exportfs -v

/nfs/vol1      10.38.12.0/24(rw,wdelay,no_root_squash,no_subtree_check,sec=sys,rw,secure,no_root_squash,no_all_squash)

k8s 노드들에도 NFS 클라이언트를 설치해줘야 한다. 설치하지 않으면 접근 시 오류가 발생한다.

Kubernetes NFS volume mount fail with exit status 32

NODE 에 개별 노드 명을 입력해서 모든 노드에 대해 실행하자.

multipass exec $NODE -- sudo apt-get install -y nfs-common

NFS 설정은 끝났다. 동적으로 볼륨을 할당받을 수 있도록 준비하자.

Kubernetes Volumes : Static & Dynamic Provisioning

이 문서에서 dynamic provisioning 을 따라가자. dynamic provisioning 은 요청때마다 새로운 디렉토리를 만들어서 할당해 준다. 미리 할당받아 두는 static provisioning 은 마운트된 볼륨에서 데이터를 읽을 용도가 있거나 지정된 같은 디렉토리에 기록해야 하는 경우가 아니라면 필요성이 덜할 것이다. static provisioning 은 훨씬 쉬우니 문서를 참조하자.

service account 와 필요한 role 들을 정의해 준다.

vim nfs-prov-sa.yaml

kind: ServiceAccount
apiVersion: v1
metadata:
  name: nfs-pod-provisioner-sa
---
kind: ClusterRole # Role of kubernetes
apiVersion: rbac.authorization.k8s.io/v1 # auth API
metadata:
  name: nfs-provisioner-clusterRole
rules:
  - apiGroups: [""] # rules on persistentvolumes
    resources: ["persistentvolumes"]
    verbs: ["get", "list", "watch", "create", "delete"]
  - apiGroups: [""]
    resources: ["persistentvolumeclaims"]
    verbs: ["get", "list", "watch", "update"]
  - apiGroups: ["storage.k8s.io"]
    resources: ["storageclasses"]
    verbs: ["get", "list", "watch"]
  - apiGroups: [""]
    resources: ["events"]
    verbs: ["create", "update", "patch"]
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: nfs-provisioner-rolebinding
subjects:
  - kind: ServiceAccount
    name: nfs-pod-provisioner-sa 
    namespace: default
roleRef: # binding cluster role to service account
  kind: ClusterRole
  name: nfs-provisioner-clusterRole # name defined in clusterRole
  apiGroup: rbac.authorization.k8s.io
---
kind: Role
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: nfs-pod-provisioner-otherRoles
rules:
  - apiGroups: [""]
    resources: ["endpoints"]
    verbs: ["get", "list", "watch", "create", "update", "patch"]
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: nfs-pod-provisioner-otherRoles
subjects:
  - kind: ServiceAccount
    name: nfs-pod-provisioner-sa # same as top of the file
    # replace with namespace where provisioner is deployed
    namespace: default
roleRef:
  kind: Role
  name: nfs-pod-provisioner-otherRoles
  apiGroup: rbac.authorization.k8s.io

kubectl apply -f nfs-prov-sa.yaml

다음으로 storage class 를 만들자.

vim storageclass-nfs.yaml

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: nfs-storageclass # IMPORTANT pvc needs to mention this name
provisioner: nfs-dynamic # name can be anything
parameters:
  archiveOnDelete: "false"

kubectl apply -f storageclass-nfs.yaml

위에서 provisioner 와 name 을 (적당히 수정하고) 잘 기억해 두자.

이제 pvc 가 요청될 때 매칭되는 pv 를 provisioning 해 줄 provisioner 를 실행하자.

vim pod-provision-nfs.yaml

kind: Deployment
apiVersion: apps/v1
metadata:
  name: nfs-pod-provisioner
spec:
  selector:
    matchLabels:
      app: nfs-pod-provisioner
  replicas: 1
  strategy:
    type: Recreate
  template:
    metadata:
      labels:
        app: nfs-pod-provisioner
    spec:
      serviceAccountName: nfs-pod-provisioner-sa # name of service account
      containers:
        - name: nfs-pod-provisioner
          image: quay.io/external_storage/nfs-client-provisioner:latest
          volumeMounts:
            - name: nfs-provisioner-v
              mountPath: /persistentvolumes
          env:
            - name: PROVISIONER_NAME # do not change
              value: nfs-dynamic # SAME AS PROVISIONER NAME VALUE IN STORAGECLASS
            - name: NFS_SERVER # do not change
              value: 192.168.1.60 # Ip of the NFS SERVER
            - name: NFS_PATH # do not change
              value: /nfs/vol1  # path to nfs directory setup
      volumes:
       - name: nfs-provisioner-v # same as volumemouts name
         nfs:
           server: 192.168.1.60
           path: /nfs/vol1

kubectl apply -f pod-provision-nfs.yaml

주석 처리된 부분을 잘 보고, provisioner 이름, NFS server 의 IP, NFS path 를 수정해 준 다음 provisioner 를 띄워 준다.

문서에는 테스트로 pvc 를 만들어서 실제로 볼륨을 할당받아보는 것까지 있는데, 문서 따라가서 테스트해 보았다고 가정하고, Spark 에서 이를 활용해 보자.

apache/spark

(대응되는 기능이 Spark 3.1.0 에 추가되어서 공식 사이트에 문서가 아직 없다. 위의 링크는 branch-3.1 의 md 파일)

약간 가져와 보자면…

For example, you can mount a dynamically-created persistent volume claim per executor by using OnDemand as a claim name and storageClass and sizeLimit options like the following. This is useful in case of Dynamic Allocation.

spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.claimName=OnDemand
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.storageClass=gp
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.sizeLimit=500Gi
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.mount.path=/data
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.mount.readOnly=false

claimName 에 “OnDemand” 라고 쓰면 Spark 가 unique 한 이름으로 대체해 준다. 중요한 부분은 storageClass 와 sizeLimit… 둘 다 명시가 되어야 한다. storageClass 는 아까 NFS 동적 프로비저닝 설정할 때 쓴 storageClass 를 쓰면 된다. mount.path 는 container 에 mount 될 경로를 지정한다. readOnly 는 그 자체가 설명이니…

개별 executor 에 적용되며, “executor” 를 “driver” 로 교체하여 driver 에서도 볼륨을 받을 수 있다. “data” 자리에 볼륨 이름이 들어가는데 볼륨이 설정 단위이기 때문에 여러 볼륨을 할당받아야 한다면 볼륨 이름을 다르게 해서 여러 개를 설정할 수 있을 것으로 보인다. (하나의 큰 볼륨을 받을 수 있다면 하나만 받아서 디렉토리로 구분해서 써도 문제가 없겠지만…)

문서에 바로 이어서 설명되어 있지만, local dir 도 볼륨을 받아서 쓸 수 있다. 볼륨 이름이 “spark-local-dir-” 로 시작하면 된다. 마운트된 경로들이 SPARK_LOCAL_DIRS 에 설정된다고 한다. 설정 방법은 볼륨 이름 부분 외에는 전부 같다. driver/executor 모두 설정 가능하다는 부분도 같다.

예를 들어, 이전 문서에서 magic committer 를 사용하는 batch query 실행 예시를 아래와 같이 변경하여 동적으로 프로비저닝된 볼륨을 활용할 수 있다.

./bin/spark-submit \
--master k8s://https://10.38.12.80:6443 \
--deploy-mode cluster \
--conf spark.executor.instances=2 \
--conf spark.kubernetes.container.image=10.38.12.80:5000/spark-py:spark-3.1.0-SNAPSHOT-bin-hadoop3.2-with-hadoop-cloud-20201223-v1 \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.stagingvol.options.claimName=OnDemand \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.stagingvol.options.storageClass=nfs-storageclass \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.stagingvol.options.sizeLimit=1Gi \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.stagingvol.mount.path=/home/spark/work-dir \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.stagingvol.mount.readOnly=false \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.spark-local-dir-1.options.claimName=OnDemand \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.spark-local-dir-1.options.storageClass=nfs-storageclass \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.spark-local-dir-1.options.sizeLimit=1Gi \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.spark-local-dir-1.mount.path=/home/spark/local-dir \
--conf spark.kubernetes.driver.volumes.persistentVolumeClaim.spark-local-dir-1.mount.readOnly=false \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.stagingvol.options.claimName=OnDemand \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.stagingvol.options.storageClass=nfs-storageclass \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.stagingvol.options.sizeLimit=1Gi \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.stagingvol.mount.path=/home/spark/work-dir \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.stagingvol.mount.readOnly=false \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.claimName=OnDemand \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.storageClass=nfs-storageclass \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.sizeLimit=1Gi \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.path=/home/spark/local-dir \
--conf spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.readOnly=false \
--conf spark.kubernetes.pyspark.pythonVersion=3 \
--conf spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a=org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory \
--conf spark.hadoop.fs.s3a.committer.name=magic \
--conf spark.hadoop.fs.s3a.committer.magic.enabled=true \
--conf spark.hadoop.fs.s3a.buffer.dir=/home/spark/work-dir/s3a-buffer \
--conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
--conf spark.kubernetes.file.upload.path=s3a://spark-upload \
--conf spark.hadoop.fs.s3a.access.key="heartsavior" \
--conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
--conf spark.hadoop.fs.s3a.fast.upload=true \
--conf spark.hadoop.fs.s3a.secret.key="heartsavior" \
--conf spark.hadoop.fs.s3a.path.style.access=true \
--conf spark.hadoop.fs.s3a.endpoint=http://10.38.12.80:9000 \
--conf spark.hadoop.fs.s3a.connection.ssl.enabled=false \
--conf spark.sql.sources.commitProtocolClass=org.apache.spark.internal.io.cloud.PathOutputCommitProtocol \
--conf spark.sql.parquet.output.committer.class=org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter \
--conf spark.driver.extraJavaOptions="-Divy.cache.dir=/tmp/ivy-cache -Divy.home=/tmp/ivy" \
./test_batch.py s3a://batch-output/query-output-a

위에서 spark.kubernetes.driver.volumes.persistentVolumeClaim 와 spark.kubernetes.executor.volumes.persistentVolumeClaim 으로 시작하는 설정들을 눈여겨 보고, 마운트된 디렉토리를 어느 설정에서 활용하는지 확인해 보자.

여러 Spark 어플리케이션에서 기록하는 event log 들을 SHS 로 서빙하는 건 테스트가 좀 필요할 것 같다. static provisioning 을 통해 하나의 PVC 를 만들고 같은 NFS 디렉토리에 기록하도록 하는 방식이 쉬워 보이긴 하는데 하나의 PVC 를 동시에 접근하고 기록하고 돌려써야 되는데 테스트가 필요해 보인다. TODO 로 남겨 두자. dynamic provisioning 도 claimName 을 같게 한다면 (OnDemand 제외) 같은 NFS 디렉토리에 기록이 될 테니 비슷하게 할 수 있겠다. 다만 처음 요청 때 디렉토리가 동적으로 생성될 거라서 그걸 신경써 줘야 한다.

개발 환경 셋업의 마무리로 k8s 에서 Structured Streaming 쿼리로 결과를 파일 (S3), Kafka 등에 기록하는 것을 테스트했었다. 많은 TODO 를 남겼지만 일단 정상 동작하는 것을 확인해서 Spark 은 얼추 실행 테스트를 마쳤구나 생각했는데, 배치 쿼리는 파일 싱크에 기록할 때 기록 방법이 조금 다르다는 것을 나중에 깨달았다.

Structured Streaming 파일 싱크는 아웃풋 디렉토리 아래에 메타데이터 디렉토리를 만들고 거기에 각 마이크로 배치 별로 기록이 정상적으로 완료된 파일들의 리스트를 파일에 기입한다. 개별 task 가 아웃풋 파일들을 쓰고, task 가 정상적으로 완료되면 driver 로 파일 리스트를 넘기고 driver 가 모든 task 에 대해 정상 완료를 수신하면 전체 파일 리스트를 기록하는 방식으로 작동한다.

다른 쿼리에서 해당 아웃풋 디렉토리를 읽을 때, 메타데이터 디렉토리에 있는 파일들을 읽어서 실제 아웃풋 파일 리스트를 뽑고, 실제 파일들을 읽게 되어 있다. End-to-end exactly-once 보장을 위한 방법인데, Spark 만 메타데이터 디렉토리를 인지할 수 있기 때문에 Spark 으로 읽어야 기록하다 실패한 파일이라던지 여타 문제 있는 파일들을 제외하고 읽을 수 있다. (메타데이터 디렉토리를 무시하고 아웃풋 디렉토리를 읽으면 at-least-once 이고 깨진 파일들도 존재할 수 있어 엔진이 오류를 무시할 수도 있어야 한다.)

배치의 파일 싱크 기록은 일반적으로 지정된 임시 디렉토리에 개별 task 가 아웃풋 파일들을 쓰고 모든 task 가 정상 완료되면 driver 가 그 임시 디렉토리를 최종 디렉토리로 옮기는 방식으로 진행된다. (file output committer algorithm version 1) version 2 는 driver 가 개입하지 않고 개별 task 가 완료될 때 자신이 작성한 아웃풋 파일들을 최종 디렉토리로 옮기는 것으로 알려져 있다. 다른 말로 하면, 몇몇 task 들이 먼저 성공하고 특정 task 가 실패하면 최종 디렉토리에 “일부” 아웃풋 파일들이 노출되고 전체 쿼리는 실패한다는 뜻이다. 아웃풋 디렉토리를 바로 읽어들이는 쿼리가 동작 중이라면 이 “불완전한” 아웃풋을 읽어 간다는 이야기. 그래서 Spark 은 Hadoop 이 version 2 를 기본 설정으로 함에도 별도 기본값을 version 1 로 유지하고 사용자가 덮어쓸 수 있게 하고 있다. (3.0.2/3.1.0 에 포함 예정이고 이전 버전들은 Hadoop 기본 설정을 따라가므로 주의!)

Use spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=1 by default

이건 HDFS 호환 파일 시스템일 때의 이야기이다. version 1 은 디렉토리 rename 이 atomic 인 파일 시스템에서만 문제 없이 작동한다. 현재 셋업은 HDFS 가 구성되어 있지 않고 구성할 생각도 없다. S3 에 기록하는 게 목표이다. (정확히는 MinIO 겠지만…) 전용 committer 가 있을까?

있다. S3A committer.

Apache Hadoop Amazon Web Services support - Committing work to S3 with the S3A Committers

이것도 크게 보면 세 종류 (file, staging, magic) 인데 file committer 는 위의 그 file output committer 같으니 무시하고, staging 은 클러스터 파일시스템을 필요로 한다. (예를 들어 HDFS) 얼추 대충 보니 클러스터 파일시스템에 먼저 기록하고 모든 기록이 성공하면 S3 로 업로드 하는 것으로 보인다.

HDFS 없이 S3 에 기록하는 게 목표라면, magic committer 외에는 선택의 여지가 없다. magic committer 는 파일시스템이 strong consistent 해야 정상 동작하고 그래서 S3 Guard 를 필수로 사용해야 한다고 명시되어 있는데, 아마존의 올해 발표로 S3 Guard 는 사실상 관짝에 들어가게 되었다. S3 호환 파일 시스템들이 strong consistent 한지는 몰라도 S3 자체는 magic committer 를 사용할 수 있는 것으로 보인다.

• Amazon S3 Update - Strong Read-After-Write Consistency | Amazon Web Services
• AWS Consistent S3 & Apache Hadoop's S3A connector

(이름이 magic 인 만큼 어떻게 동작하는 지 문서에도 딱히 설명된 게 없다… 디테일은 나중에 한 번 찾아보기로 하고…)

magic committer 를 사용하도록 설정해서 아래의 아주 단순한 배치 쿼리를 실행해 보자.

import sys
from pyspark.sql import SparkSession

def main(output_path):
    spark.range(100000).repartition(100).write.json(output_path)
    df = spark.read.json(output_path)
    df.show()

if __name__ == "__main__":
    spark = SparkSession \
        .builder \
        .appName("Python Spark SQL test") \
        .getOrCreate()

path = sys.argv[1]
print("path: %s" % path)

main(path)

S3A committer 를 사용하려면 spark-hadoop-cloud 모듈에 있는 클래스를 사용해야 하는데 내가 아는 게 맞다면 해당 모듈은 (바뀔지 모르겠지만) 기본 프로파일로는 포함되지 않는다. 다시 말해서, 현재 시점에서는 직접 빌드를 해야 된다는 뜻이다.

Integration with Cloud Infrastructures

어차피 직접 빌드해야 되는 마당이니 3.1 브랜치를 빌드하자. (k8s 지원은 Spark 3.1 에서 GA 가 되기 때문에 아직 릴리즈 되지 않았어도 3.1 로 테스트하는 게 사서 고생을 덜할 수도 있다.)

git clone https://github.com/apache/spark.git
git checkout branch-3.1
mvn clean install -DskipTests -Pkubernetes -Phadoop-3.2 -Phadoop-cloud
dev/make-distribution.sh -Pkubernetes -Phadoop-3.2 -Phadoop-cloud

dist 디렉토리를 적당히 복사하고, 해당 디렉토리로 이동한다.

그리고 docker image 를 만들고 private docker registry 에 밀어넣자.

sudo ./bin/docker-image-tool.sh -r 10.38.12.80:5000 -t spark-3.1.0-SNAPSHOT-bin-hadoop3.2-with-hadoop-cloud-20201223-v1 -p ./kubernetes/dockerfiles/spark/bindings/python/Dockerfile build

sudo ./bin/docker-image-tool.sh -r 10.38.12.80:5000 -t spark-3.1.0-SNAPSHOT-bin-hadoop3.2-with-hadoop-cloud-20201223-v1 -p ./kubernetes/dockerfiles/spark/bindings/python/Dockerfile push

hadoop-cloud 프로파일은 hadoop-aws 와 transitive dependencies (amazon-sdk 등등) 들을 포함하므로 더 이상 packages 란에 넣어 주지 않아도 된다. (처음부터 이렇게 했었어야 했나보다…)

bounty castle 관련 라이브러리가 없다는 에러는 여전히 발생한다. submit 할 때에만 있어도 되는 것으로 보인다. 예전 문서에서 언급한 것 처럼 다운로드 받아서 넣어준다. Spark docker image 에는 포함되지 않아도 되는 것 같다.

실행은 아래와 같이 한다. (test_batch.py 는 위의 배치 쿼리 코드가 기록된 파일)

./bin/spark-submit \
--master k8s://https://10.38.12.80:6443 \
--deploy-mode cluster \
--conf spark.executor.instances=2 \
--conf spark.kubernetes.container.image=10.38.12.80:5000/spark-py:spark-3.1.0-SNAPSHOT-bin-hadoop3.2-with-hadoop-cloud-20201223-v1 \
--conf spark.kubernetes.pyspark.pythonVersion=3 \
--conf spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a=org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory \
--conf spark.hadoop.fs.s3a.committer.name=magic \
--conf spark.hadoop.fs.s3a.committer.magic.enabled=true \
--conf spark.hadoop.fs.s3a.buffer.dir=/opt/spark/work-dir/s3a-buffer \
--conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
--conf spark.kubernetes.file.upload.path=s3a://spark-upload \
--conf spark.hadoop.fs.s3a.access.key="heartsavior" \
--conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
--conf spark.hadoop.fs.s3a.fast.upload=true \
--conf spark.hadoop.fs.s3a.secret.key="heartsavior" \
--conf spark.hadoop.fs.s3a.path.style.access=true \
--conf spark.hadoop.fs.s3a.endpoint=http://10.38.12.80:9000 \
--conf spark.hadoop.fs.s3a.connection.ssl.enabled=false \
--conf spark.sql.sources.commitProtocolClass=org.apache.spark.internal.io.cloud.PathOutputCommitProtocol \
--conf spark.sql.parquet.output.committer.class=org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter \
--conf spark.driver.extraJavaOptions="-Divy.cache.dir=/tmp/ivy-cache -Divy.home=/tmp/ivy" \
./test_batch.py s3a://batch-output/query-output-1

/opt/spark/work-dir 은 Spark docker image 가 생성하는 디렉토리이다. 일단 이 공간을 사용하되, 나중에는 NFS 에서 동적으로 할당받아서 쓰려고 한다. (k8s 노드에 디스크 공간 할애를 최대한 적게 하고, NFS 를 통해 호스트 서버의 디스크를 무작위로 사용하게 하려고 함.)

magic committer 사용 관련 변경이 있는 설정들만 다시 모아 보면…

--conf spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a=org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory \
--conf spark.hadoop.fs.s3a.committer.name=magic \
--conf spark.hadoop.fs.s3a.committer.magic.enabled=true \
--conf spark.hadoop.fs.s3a.buffer.dir=/opt/spark/work-dir/s3a-buffer \
--conf spark.sql.sources.commitProtocolClass=org.apache.spark.internal.io.cloud.PathOutputCommitProtocol \
--conf spark.sql.parquet.output.committer.class=org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter \

이 정도이다. 하둡 설정 관련은 아래 문서를 참조하고…

Apache Hadoop Amazon Web Services support - Committing work to S3 with the S3A Committers

Spark 설정 관련은 아래 문서를 참조했다.

Integration with Cloud Infrastructures

실제로 실행해 보면, 아웃풋 디렉토리에 아래와 같이 파일들이 생성된다.

mc ls minio-k3s/batch-output/query-output-8/
[2020-12-23 17:53:03 KST]  12KiB _SUCCESS
[2020-12-23 17:52:54 KST]  13KiB part-00000-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:52:54 KST]  13KiB part-00001-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:52:55 KST]  13KiB part-00002-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:52:55 KST]  13KiB part-00003-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:52:55 KST]  13KiB part-00004-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:52:55 KST]  13KiB part-00005-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
...
[2020-12-23 17:53:03 KST]  13KiB part-00095-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:53:03 KST]  13KiB part-00096-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:53:03 KST]  13KiB part-00097-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:53:03 KST]  13KiB part-00098-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json
[2020-12-23 17:53:03 KST]  13KiB part-00099-0d07120a-af31-46d8-8649-bdada9dd4246-c000.json

_SUCCESS 파일에 아웃풋에 대한 메타데이터가 기록되어 있다. 한 번 직접 확인해 보자.

mc cat minio-k3s/batch-output/query-output-8/_SUCCESS

이로써 magic committer 를 사용하면 k8s 노드들의 로컬 파일 시스템만 활용해서 Spark 를 통해 S3 에 파일 기록이 가능함을 확인할 수 있었다.

이제 k3s 클러스터에 MinIO 도 설치되었고 private docker registry 도 설정되었다. 이제 k3s 클러스터에 Spark 어플리케이션을 실행해 보자. (나중에는 remote shuffle service 도 설정해보고 할 예정인데 아직은 k8s 자체도 잘 모르니 쉬운 것부터…)

최신 버전인 Spark 3.0.1 의 바이너리를 다운로드받고 (spark-3.0.1-hadoop-3.2 로 사용함) <SPARK_HOME> 이라고 하자.

실행해 보니 bounty castle 관련 라이브러리가 없다는 에러가 나타나서 다운로드 받아서 넣어주는 것으로 했다. Spark docker image 를 다시 만들때에도 포함이 된다.

cd <SPARK_HOME>/jars

wget https://repo1.maven.org/maven2/org/bouncycastle/bcpkix-jdk15on/1.67/bcpkix-jdk15on-1.67.jar
wget https://repo1.maven.org/maven2/org/bouncycastle/bcprov-jdk15on/1.67/bcprov-jdk15on-1.67.jar

이제 image 를 빌드하고 push 해 보자. (pyspark 을 사용할 예정이라 해당 Dockerfile 을 사용하였다. Spark k8s 가이드 문서에 명시되어 있다.)

이전 문서에서 설정한 대로, http://10.38.12.80:5000 이 private docker registry 의 endpoint 라고 하자.

sudo ./bin/docker-image-tool.sh -r 10.38.12.80:5000 -t spark-3.0.1-bin-hadoop3.2-with-bounty-castle-20201221-v1 -p ./kubernetes/dockerfiles/spark/bindings/python/Dockerfile build

sudo ./bin/docker-image-tool.sh -r 10.38.12.80:5000 -t spark-3.0.1-bin-hadoop3.2-with-bounty-castle-20201221-v1 -p ./kubernetes/dockerfiles/spark/bindings/python/Dockerfile push

private docker registry 에 image 가 push 되었다.

실행하기 전에, service account 를 만들고 cluster role binding 을 해당 account 에 설정해 Spark driver pod 이 executor pod 을 띄울 수 있게 해 준다.

kubectl create serviceaccount spark
kubectl create clusterrolebinding spark-role --clusterrole=edit --serviceaccount=default:spark --namespace=default

그리고 private docker registry 에 인증할 수 있게 secret 을 만들어 준다.

kubectl create secret docker-registry regcred --docker-server=http://10.38.12.80:5000 --docker-username="<docker registry username>" --docker-password="<docker registry password>" --docker-email="<email address>"

Pull an Image from a Private Registry

여기서 만든 regcred 는 imagePullSecrets 에 제시되어야 한다. Spark 은 해당 설정을 노출하고 있어서 옵션으로 넣어주면 된다.

아래는 SparkPi 를 실행하는 명령이다. 실행 전에 mc mb minio-k3s/spark-upload 로 필요한 bucket 을 생성하는 것도 잊지 말자.

• K8S API endpoint : https://10.38.12.80:6443
• MinIO endpoint : http://10.38.12.116:9000
• private docker registry endpoint : http://10.38.12.80:5000

<SPARK_HOME>/bin/spark-submit \
    --master k8s://https://10.38.12.80:6443 \
    --deploy-mode cluster \
    --name spark-pi \
    --class org.apache.spark.examples.SparkPi \
    --conf spark.executor.instances=1 \
    --conf spark.kubernetes.container.image=10.38.12.80:5000/spark:spark-3.0.1-bin-hadoop3.2-with-bounty-castle-20201221-v1 \
    --conf spark.kubernetes.container.image.pullSecrets=regcred \
    --conf spark.kubernetes.local.dirs.tmpfs=true \
    --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
    --packages org.apache.hadoop:hadoop-aws:3.2.0 \
    --repositories https://repo1.maven.org/maven2/ \
    --conf spark.kubernetes.file.upload.path=s3a://spark-upload \
    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
    --conf spark.hadoop.fs.s3a.access.key="heartsavior" \
    --conf spark.hadoop.fs.s3a.secret.key="heartsavior" \
    --conf spark.hadoop.fs.s3a.fast.upload=true \
 --conf spark.hadoop.fs.s3a.path.style.access=true \
    --conf spark.hadoop.fs.s3a.endpoint=http://10.38.12.116:9000 \
    --conf spark.hadoop.fs.s3a.connection.ssl.enabled=false \
    --conf spark.driver.extraJavaOptions="-Divy.cache.dir=/tmp -Divy.home=/tmp" \
    file://<SPARK_HOME>/examples/jars/spark-examples_2.12-3.0.1.jar

실행하면 Yarn 에 Spark application 을 제출할 때 처럼 pending 이 지속되다가 running 혹은 error 가 발생하고 application 이 끝날 때까지 멈추지 않는다. 다른 콘솔 창을 띄워서 k9s 를 실행하고 pods 를 지켜보면 driver 가 실행되고, 이어 executor 가 실행된다. driver pod 을 kill 하면 executor pod 은 자연 종료된다. Spark PI 는 계산이 끝나면 종료되고, 상태는 completed 로 남는다. pod 이 자동으로 사라지지 않아 로그를 확인하거나 설정 등을 차후에 확인할 수 있지만, 실행 후에는 지워주는 게 좋겠다.

이제 기본적인 Spark application 실행이 가능해졌다. jar 로 실행하는 어플리케이션은 그냥 이 틀에 맞춰 실행해도 될 것이다. 최종 목표는 Structured Streaming query 를 실행하는 거였고 내가 갖고 있는 쿼리들은 PySpark 으로 구현되어 있으므로 거기에 맞게 명령을 조금 바꾸기로 한다.

그 전에, 기왕에 Structured Streaming query 를 실행하는 거니까 Kafka 도 활용해 보려고 한다. 그래봐야 데탑 한 대에 전부 띄우는 거고 적당히 동작하는지 확인하는 용도니까 브로커 1개짜리 최소 설치로 진행한다. 그래도 k8s 클러스터가 있으니 Kafka 도 k8s 에 올려보기로 하자.

역시 쉽고 편하게 간다. strimzi 의 kafka operator 를 사용한다.

https://strimzi.io/quickstarts/

kubectl create namespace kafka
kubectl apply -f '<https://strimzi.io/install/latest?namespace=kafka>' -n kafka
kubectl apply -f <https://strimzi.io/examples/latest/kafka/kafka-persistent-single.yaml> -n kafka

이대로 설치하면 ClusterIP 로 bootstrap address 가 노출된다.

kubectl get service -w my-cluster-kafka-bootstrap -n kafka
NAME                         TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)                      AGE
my-cluster-kafka-bootstrap   ClusterIP   10.43.152.51   <none>        9091/TCP,9092/TCP,9093/TCP   5d

호스트 서버에서도 사용할 수 있게 external IP 를 따서 쓰고 싶지만 external IP 를 이용해 bootstrap address 를 지정해도 결국 broker advertised host/port 를 사용하게 되어 접근이 불가능해진다. Kafka 를 제대로 k8s 에서 돌리는 게 목적은 아니니까 이 정도에서 그치고 넘어가자.

HeartSaVioR/structured_streaming_experiments

확인한 걸로는 Spark 3.0.1 의 PySpark 은 cluster mode 에서 driver 의 PYTHONPATH 가 py-files 를 반영하지 못하는 것으로 보였다. branch-3.1 을 직접 빌드해서 테스트해 보니 잘 되는 것 같다. PySpark 부분은 직접 빌드한 3.1.0-SNAPSHOT 으로 진행하자.

git clone https://github.com/apache/spark.git
git checkout branch-3.1
mvn clean install -DskipTests -Pkubernetes -Phadoop-3.2
dev/make-distribution.sh -Pkubernetes -Phadoop-3.2

여기서 dist 디렉토리가 생성되는데 이 디렉토리를 적당한 위치로 옮기자. (향후에는 SPARK_3.1.0_SNAPSHOT_HOME 으로 지칭)

위에서 했던 것과 마찬가지로 bounty castle 의 jar 들을 받아서 jars 디렉토리에 넣어 주고, image 를 build 하고 push 해 주자. (tag 는 조금 바꿔주는 게 좋겠다. 아래에서는 spark-3.1.0-bin-hadoop3.2-SNAPSHOT-with-bounty-castle-20201221-v1 로 설정한 것으로 간주)

내가 링크한 Structured Streaming 프로젝트는 clone 받아서 make clean -> make all 하면 dist 디렉토리에 파일들이 생성된다. Python 3.6 에서 테스트했다.

프로젝트 디렉토리에서 아래의 명령을 실행하자. (checkpoint 도 S3 에 넣도록 설정할 예정인데 bucket 을 따로 파서 사용하고자 한다. MinIO 에 ‘spark-checkpoints’ bucket 을 만들어 주자.)

<SPARK_3.1.0_SNAPSHOT_HOME>/bin/spark-submit \
    --master k8s://https://10.38.12.80:6443 \
    --deploy-mode cluster \
    --py-files ./dist/jobs.zip,./dist/libs.zip \
    --conf spark.executor.instances=1 \
    --conf spark.kubernetes.container.image=10.38.12.80:5000/spark-py:spark-3.1.0-bin-hadoop3.2-SNAPSHOT-with-bounty-castle-20201221-v1 \
    --conf spark.kubernetes.local.dirs.tmpfs=true \
    --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
    --jars /home/heartsavior/.m2/repository/org/apache/spark/spark-sql-kafka-0-10_2.12/3.1.0-SNAPSHOT/spark-sql-kafka-0-10_2.12-3.1.0-SNAPSHOT.jar,/home/heartsavior/.m2/repository/org/apache/spark/spark-token-provider-kafka-0-10_2.12/3.1.0-SNAPSHOT/spark-token-provider-kafka-0-10_2.12-3.1.0-SNAPSHOT.jar,/home/heartsavior/.m2/repository/org/apache/kafka/kafka_2.12/2.6.0/kafka_2.12-2.6.0.jar,/home/heartsavior/.m2/repository/org/apache/kafka/kafka-clients/2.6.0/kafka-clients-2.6.0.jar,/home/heartsavior/.m2/repository/org/apache/commons/commons-pool2/2.6.2/commons-pool2-2.6.2.jar \
    --packages org.apache.hadoop:hadoop-aws:3.2.0 \
    --repositories https://repo1.maven.org/maven2/ \
    --conf spark.kubernetes.file.upload.path=s3a://spark-upload \
    --conf spark.hadoop.fs.s3a.access.key="heartsavior" \
    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
    --conf spark.hadoop.fs.s3a.fast.upload=true \
    --conf spark.hadoop.fs.s3a.secret.key="heartsavior" \
    --conf spark.hadoop.fs.s3a.path.style.access=true \
    --conf spark.hadoop.fs.s3a.endpoint=http://10.38.12.116:9000 \
    --conf spark.hadoop.fs.s3a.connection.ssl.enabled=false \
    --conf spark.driver.extraJavaOptions="-Divy.cache.dir=/tmp -Divy.home=/tmp" \
    --conf spark.kubernetes.pyspark.pythonVersion=3 \
    file://$PWD/dist/main.py \
    simple_query --source-format=rate --sink-format=kafka \
    --source-option-rows-per-second=100 \
    --sink-option-bootstrap-servers=10.43.152.51:9092 \
    --sink-option-topic="ss-output" \
    --checkpoint-location=s3a://spark-checkpoints/checkpoint-query-1

SNAPSHOT 빌드를 사용하므로 packages 로 spark-sql-kafka 를 지정할 수 없어서 jars 에 transitive dependencies 를 모두 넘겨 주어야 한다. mvn install 를 먼저 실행했으므로 jar 파일들은 .m2 에 이미 마련되어 있다. 찾아서 jars 옵션에 넣어 준다.

잘 실행이 됐다면 Kafka 에 ss-output topic 이 만들어지고 데이터가 유입되고 있을 것이다.

여기까지가 몇 주 간의 삽질 일기이다. 대충 넘긴 게 많은 만큼 아직 많은 TODO 가 남아 있다. 스토리지 쪽은 MinIO 에 완전히 미뤄놓았지만 막상 MinIO 는 기본 스토리지 설정을 사용하고 있는데 이게 pod 이 실행된 노드에 500G 를 요구하는 것 같다. (없다고 실행이 안되는 것도 아니지만, 그래도 그냥 찜찜…) private docker registry 는 기본적으로 TLS 설정을 하는 만큼 결과적으로는 TLS 설정을 하고 workaround 들을 치워내는 게 좋겠고, 이왕 한 김에 MinIO 도 TLS 설정을 하면 좋겠다. Kafka 도 broker 의 advertised host/port 를 설정해서 호스트 서버에서 접근 가능하게 하면 좋겠고… Spark 도 규모가 있는 쿼리는 spark.kubernetes.local.dirs.tmpfs=true 를 치우고 PV 를 붙여줘야 될 것 같고…

TODO 가 해결되면 해결되는 대로 문서를 업데이트하거나 추가 문서를 또 올릴 생각이다. 그 전에 k8s 자체에 대한 학습을 먼저 할 예정이지만…

위의 문서를 통해 k3s 클러스터의 기본 설정이 완료되었다. 추가 설정 없이 docker hub 를 통해 이미지를 pull 하고 push 할 수 있지만, 여러 제약도 있고 리모트 서버에 읽고 써야 되어서 아무래도 빠릿빠릿한 부분이 덜하다. 그래서 많은 경우 private docker registry 를 구성하고 이를 사용하도록 하는 듯 하다.

docker registry 를 구성하려면 역시 스토리지를 설정해야 하는데, docker registry helm chart 에서 S3 에 저장하는 기능을 제공한다. 물론 private docker registry 를 k8s 클러스터에서 사용하는데 실제 S3 를 다녀와야 되면 비용도 시간도 낭비니까 우리는 MinIO 라는 훌륭한 제품으로 대체한다.
(S3 는 Spark on K8S 에서는 접근성이 괜찮은 스토리지로 볼 수 있으니 설치해두면 이래저래 쓸모가 있다.)

minio/charts

공부해가면서 설치하는 게 목적이 아니니까 helm chart 로 빠르게 설치해 보자. 일단 helm 을 설치하고…

sudo snap install helm --classic

minio 네임스페이스를 만들고 해당 네임스페이스에 설치하기로 한다.

kubectl create namespace minio

helm install minio-release minio/minio --set accessKey=<accesskey>,secretKey=<secretkey>,service.type=LoadBalancer,service.port=9000, --namespace minio --atomic

호스트 서버에서도 MinIO 에 접근할 수 있어야 되어서 service type 은 Load Balancer 로 설정했다. (이렇게 하면 external IP 가 발급되고 호스트 서버에서 접근 가능하다. 기본값은 ClusterIP 이고 internal IP 만 발급되는데 k8s 클러스터 내에서만 접근 가능하다.)

accessKey 와 secretKey 값은 잘 보관하도록 하자. 접근할 때 필요하다.

이전 문서에서 언급했지만 pod 이 4G 노드를 필요로 한다. 워커 노드들의 메모리 설정을 확인하고 안전하게 메모리 5G 정도로 노드 하나를 준비하도록 하자.

실제로 사용하려면 external IP 와 port 를 알아야 한다. 아래 명령을 실행하자.

kubectl get service -w minio-release -n minio

NAME            TYPE           CLUSTER-IP    EXTERNAL-IP    PORT(S)          AGE
minio-release   LoadBalancer   10.43.51.27   10.38.12.116   9000:31310/TCP   6h23m

위의 결과에서 endpoint URL 은 http://10.38.12.116:9000 이다.

별도로 MinIO CLI 를 호스트 서버에 설치해야 한다. 손쉽게 바이너리를 바로 다운받아 쓰자.

wget https://dl.min.io/client/mc/release/linux-amd64/mc
chmod +x mc
./mc --autocompletion

이 mc 파일은 PATH 에 포함되는 위치로 잘 옮겨 준다.

다음, k8s 에 설치된 MinIO 에 접근할 수 있게 alias 를 걸어 주고, private docker registry 에 사용할 bucket 을 생성하자.

mc alias set minio-k3s http://10.38.12.116:9000 "<access key>" "<secret key>" --api s3v4
mc mb minio-k3s/docker-registry
mc ls minio-k3s

이제 MinIO 는 설정이 완료되었다. 본격적으로 private docker registry 를 설정해보자.

사실 TLS 설정을 하는 게 정석인 듯 한데 생각보다 잘 안되는 부분이 많아서 HTTP 로 설정했다. docker 나 containerd 모두 HTTPS 를 요구하기 때문에 별도 설정이 필요한데, 이 쪽이 삽질이 덜하긴 할 것 같다. TLS 설정이 잘 되면 HTTP 를 사용하기 위한 별도 설정은 제거해도 된다.

모든 문서에서 htpasswd 를 통한 아주 기초적인 인증은 기본적으로 설정하는 듯 해서 여기서도 한다. (htpasswd 를 사용하려면 apache2-utils 를 설치하거나 htpasswd 가 설치된 docker image 를 통해 실행해야 하는데 그냥 설치한다.)

sudo apt-get install apache2-utils
htpasswd -Bbn <docker registry username> <docker registry password> > ./htpasswd

kubectl create namespace docker-registry

helm install registry stable/docker-registry \
--set s3.region=us-east-1 \
--set s3.regionEndpoint=http://10.38.12.116:9000 \
--set s3.secure=false \
--set s3.bucket=docker-registry \
--set secrets.s3.accessKey=<minio access key> \
--set secrets.s3.secretKey=<minio secret key> \
--set secrets.htpasswd=$(cat ./htpasswd) \
--set storage=s3 \
--set service.type=LoadBalancer \
--namespace docker-registry

마찬가지로 실제로 사용하려면 external IP 와 port 를 알아야 한다. 아래 명령을 실행하자.

kubectl get svc -w registry-docker-registry -n docker-registry
NAME                       TYPE           CLUSTER-IP     EXTERNAL-IP   PORT(S)          AGE
registry-docker-registry   LoadBalancer   10.43.93.191   10.38.12.80   5000:30968/TCP   5h54m

위의 결과에서 endpoint URL 은 http://10.38.12.80:5000 이다. TLS 를 설정하지 않았다는 점을 염두에 두자.

docker 와 containerd 에서 http endpoint URL 를 허용하도록 설정을 좀 해야 된다.

그 전에, docker 를 호스트 서버에 설치하지 않았다면 설치해 주어야 한다. linuxbrew 로 설치하면 docker client 만 설치되므로, snap 을 활용해서 설치해 준다. 실행할 때마다 sudo 를 필요로 하는 불편함이 있지만 나중에 적당히 alias 를 걸어주자.

sudo snap install docker

설치가 되었다면, 아래와 같이 설정을 변경한다. snap 이 아닌 방법으로 설치했다면 설치 방법에 따라 설정 파일 위치가 다를 것이다. (구글링)

sudo vim /var/snap/docker/current/config/daemon.json

json 내용에 아래 내용을 추가한다.

"insecure-registries":["10.38.12.80:5000"]

그리고 docker daemon 을 재시작한다.

sudo snap restart docker

아래 명령을 실행했을 때,

sudo docker info

...

 Insecure Registries:
  10.38.12.80:5000
  127.0.0.0/8

가 나오면 제대로 설정된 것이다. 로그인해보자.

sudo docker login http://10.38.12.80:5000 -u <docker registry username>

프롬프트에 docker registry password 로 설정한 값을 입력한다. Login Succeeded 라는 메시지가 나타나야 한다.

몇 가지 테스트를 더 해 보자.

export SERVER="10.38.12.80:5000"
sudo docker pull functions/figlet:latest
sudo docker tag functions/figlet:latest $SERVER/functions/figlet:latest
sudo docker push $SERVER/functions/figlet:latest

이렇게 하면 functions/figlet:latest 가 우리가 설정한 private docker registry 에 push 되어야 한다. 확인해보자.

curl -X GET http://10.38.12.80:5000/v2/_catalog -u <docker registry username> -p

프롬프트가 나타나면 패스워드를 입력한다. 결과는 아래와 같을 것이다.

{"repositories":["functions/figlet"]}

호스트 서버에서는 설정이 완료되었다. 하지만 k8s 노드들은 여전히 HTTP 를 사용하는 private docker registry 를 거부한다. 여기도 설정을 하자.

vim registries.yaml

아래와 같이 채워 넣고…

mirrors:
  "10.38.12.80:5000":
    endpoint:
      - "http://10.38.12.80:5000"
configs:
  "10.38.12.80:5000":
    auth:
      username: <docker registry username>
      password: <docker registry password>

마스터 노드와 워커 노드에 실행해야 할 명령이 살짝 다르다. 설정 파일은 같은데 리스타트해야 할 서비스 명이 다르다. 아래를 참조하자.

마스터 노드:

multipass transfer registries.yaml k3s-master:/home/ubuntu/registries.yaml
multipass exec k3s-master -- sudo cp /home/ubuntu/registries.yaml /etc/rancher/k3s/registries.yaml
multipass exec k3s-master -- sudo cat /etc/rancher/k3s/registries.yaml
multipass exec k3s-master -- sudo service k3s restart

워커 노드:

multipass transfer registries.yaml k3s-master:/home/ubuntu/registries.yaml
multipass exec k3s-master -- sudo cp /home/ubuntu/registries.yaml /etc/rancher/k3s/registries.yaml
multipass exec k3s-master -- sudo cat /etc/rancher/k3s/registries.yaml
multipass exec k3s-master -- sudo service k3s-agent restart

전체 노드에 설정해야 한다. 한꺼번에 하지 말고 천천히 한 대씩 해주자.

이제 private docker registry 에 push 한 image 를 사용하는 pod 을 여러 개 배포해서 노드들이 image 를 받는 데 문제가 없는지 확인하자.

(실제로는 Spark 관련 설정을 먼저 해서 Spark image 를 push 하고 테스트에 활용했었다. private docker registry 에 push 가 되어 있다면 어떤 image 를 사용해도 관계없어야 할 것이다.)

Kafka, S3 (MinIO), docker private repository 정도 셋업하고 이걸 활용해서 Spark 쿼리를 실행하는 것까지 우선 해 보았다. (그래서 노드가 상당히 많다. 32G 머신인데 32G 더 달아야되나 고민중…) 이번 문서에서는 k3s 셋업만 다뤄보려고 한다.

사실 삽질 과정은 지루한데 정리하면 뭐 없다. 우선 노드들을 띄우자.

multipass launch --name k3s-master --cpus 2 --mem 2048M --disk 20G
multipass launch --name k3s-node1 --cpus 1 --mem 2048M --disk 20G
multipass launch --name k3s-node2 --cpus 1 --mem 2048M --disk 20G
multipass launch --name k3s-node3 --cpus 1 --mem 2048M --disk 20G
multipass launch --name k3s-node4 --cpus 1 --mem 2048M --disk 20G
multipass launch --name k3s-minio --cpus 1 --mem 5120M --disk 100G

k3s-master 가 k3s 의 마스터로 사용할 노드이고 나머지는 워커 노드들이다. minio 노드만 메모리를 5G 로 할애했는데, minio pod 이 실행에 4G 를 요구해서 메모리를 충분히 할애했다. S3 지원하는 거면 뭐든 저기에 때려넣을 예정이라 디스크도 좀 더 줬다.

모두 실행하고 나서 아래 명령을 실행해 재확인하자.

multipass list

Name                    State             IPv4             Image
primary                 Stopped           --               Ubuntu 20.04 LTS
k3s-master              Running           10.38.12.80      Ubuntu 20.04 LTS
k3s-minio               Running           10.38.12.221     Ubuntu 20.04 LTS
k3s-node1               Running           10.38.12.210     Ubuntu 20.04 LTS
k3s-node2               Running           10.38.12.20      Ubuntu 20.04 LTS
k3s-node3               Running           10.38.12.116     Ubuntu 20.04 LTS
k3s-node4               Running           10.38.12.95      Ubuntu 20.04 LTS

multipass info --all

...

정상적으로 가상머신들이 구성되었다.

마스터 노드를 구성하자. 사실 직접 할 게 거의 없다. k3s 가 정말 편한 시스템인 듯 하다.

multipass exec k3s-master -- /bin/bash -c "curl -sfL https://get.k3s.io | K3S_KUBECONFIG_MODE="644" sh -"

설치가 완료되었으면 토큰 정보를 얻어야 한다. 워커 노드들 설치할 때 마스터 노드로 연결해 주기 위해 필요하다.

multipass exec k3s-master -- /bin/bash -c "sudo cat /var/lib/rancher/k3s/server/node-token"

출력되는 정보를 잘 복사해 두자.

모든 워커 노드들에 아래의 명령으로 k3s 를 설치한다. 여기서 K3S_TOKEN 자리에 위에서 얻은 정보를 기입하고, K3S_URL 에 마스터 노드의 IP 를 넣어준다.

multipass exec k3s-node1 -- /bin/bash -c "curl -sfL https://get.k3s.io | K3S_TOKEN=\"<토큰 정보>\" K3S_URL=https://<마스터 노드 IP>:6443 sh -"

k3s-node1 자리만 노드 이름으로 바꾸어 모두 설치한다. (스크립트 파일이던 배쉬 셸로 콘솔에서 하던 수작업하던 모두 자유)

다음으로, 호스트 서버에 Kubernetes client 가 없으면 설치해 준다.

sudo snap install kubectl --classic

(snap 이던 linuxbrew 던 직접 설치던 상관없을 것 같다.)

이제 설치한 k3s 를 호스트 서버에서 접근할 수 있는 과정을 거친다.

multipass transfer k3s-master:/etc/rancher/k3s/k3s.yaml ~/config
mkdir -p ~/.kube
mv ~/config ~/.kube/config

kubectl 이 읽는 기본 설정 파일 위치에 k3s 설정 파일을 복사한다. 이미 다른 k8s 클러스터를 접근할 수 있게 설정했다면 덮어쓰지 말고 기존 파일에 섹션 별로 나누어 복사한 다음 이름/컨텍스트들 적당히 겹치지 않게 수정하자.

k8s API 주소가 127.0.0.1 로 되어 있을 것이므로 마스터 노드 IP 로 수정해준다.

저장하고 나와서 아래의 명령을 실행한다.

kubectl get nodes

NAME         STATUS   ROLES    AGE     VERSION
k3s-node3    Ready    <none>   4d22h   v1.19.5+k3s1
k3s-minio    Ready    <none>   4d22h   v1.19.5+k3s1
k3s-master   Ready    master   4d22h   v1.19.5+k3s1
k3s-node2    Ready    <none>   4d22h   v1.19.5+k3s1
k3s-node4    Ready    <none>   2d10h   v1.19.5+k3s2
k3s-node1    Ready    <none>   4d22h   v1.19.5+k3s1

이런 식으로 출력되면 정상적으로 설치가 된 것이다. (AGE 는 훨씬 짧을 것이다. 셋업한 지 5일만에 정리하는 거라;;)

다음으로 Kubernetes 계의 mdir 인 k9s 를 설치한다. 학습하기엔 방해가 되는 유틸리티겠지만 있어야 고생을 면한다.

linuxbrew 로 설치했다. 왜 snap 으로 설치하지 않았는지는 잊어버렸다. 아마 실행할 때마다 sudo 를 필요로 했던 것 같다.

먼저 linuxbrew 설치…

/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install.sh)"
echo 'eval $(/home/linuxbrew/.linuxbrew/bin/brew shellenv)' >> /home/heartsavior/.zshrc

두번째는 zsh 쓰면 .zshrc, bash 쓰면 .bash_profile, 기타 셸은 거기에 맞게 넣어준다.

이제 k9s 설치…

brew install k9s

k9s 를 실행해서 뭔가 리스트들이 뜨면 성공이다. :nodes 했을 때 아까 kubectl get nodes 와 같은 결과가 나오면 된다. .kube/config 의 기본 컨텍스트가 설치한 k3s 이기 때문에 기본으로 k3s 클러스터의 정보가 나타나게 된다. 여러 컨텍스트들이 있다면 물론 전환해 가며 사용 가능하다. k9s 사용 방법은 구글링하거나 명령들 키가 나와 있으니 쉽게 사용 가능하다.

나의 주된 관심사 및 공헌 분야는 Structured Streaming 이라서 이 분야에 도입될 변화들만 간단하게 정리해 보려고 한다. 다른 분야들은 3.0 출시 전후로 많은 글이 올라올 것 같기도 하고 잘 모르기도 하니… JIRA 기준으로 정리했기 때문에 SQL 로 분류되었지만 Structured Streaming 에도 영향을 끼치는 변화들도 제외되었다.

(DISCLAIMER: 공헌을 직접 하다 보니 아무래도 본인의 작업들이 다수 포함되어 있다. 덜 객관적으로 소개하거나 중요한 기능들이 소개에 누락되어 있을 수 있다. JIRA 기준으로 단순 버그패치나 단순 기능개선 외에는 최대한 담아보려고 했다.)

Kafka 관련 변화가 많으니 Kafka 부터 짚어 보자.

• Kafka version upgrade

Kafka client version 이 현재 기준 2.4.0 으로 업그레이드 되었다. 하위 호환성을 잘 지원하고 있는 Kafka client 이기 때문에 하위 버전 Kafka 클러스터와도 문제가 없을 것으로 추측은 되지만 혹시 모르니 알아두면 좋을 것 같다.

Update Kafka to a verison that supports Scala 2.13

• Kafka delegation token

Kafka delegation token 을 지원한다. Source/Sink 둘 다 적용되며, 한 쿼리에 여러 Kafka cluster 를 연결하는 경우 또한 지원한다. Proxy user 는 Kafka 자체에서 아직 지원하지 않는 관계로 지원하지 않는다.

Kafka delegation token support

(외 다수)

• Header 지원

레코드의 Header 정보를 얻거나 Kafka 로 기록할 수 있게 되었다.

Source 의 경우 옵션으로 켜고 끌 수 있으며 기본적으로는 꺼져 있다. (스키마가 변경되면서 기존 stateful query 들이 영향을 받을 수 있기 때문)

Sink 의 경우 column 이 추가되어 해당 column 에 값을 채워넣으면 레코드의 Header 로 기록된다.

Add support for Kafka headers in Structured Streaming

• Kafka Source 에서 timestamp offset 지정 가능

Source 에서 읽을 범위를 지정할 때 offset 값 및 earliest/latest 외에도 timestamp 를 입력할 수 있게 되었다. 내부적으로 Kafka 의 offset by timestamp 를 호출하므로 같은 동작을 기대하면 된다. 시작 및 끝 범위 지정에 선택적으로 지정 가능하다. (끝 범위는 스트리밍 쿼리에서는 기본적으로 무시되니 주의.)

Introduce new option to Kafka source - specify timestamp to start and end offset

• Partition column 지원

선택적으로 Kafka 의 partitioner 를 사용하지 않고 row 별로 파티션 “번호" 를 직접 지정하여 Kafka 에 레코드를 기록할 수 있게 되었다.

Support partition column when writing to Kafka

• consumer group ID 의 prefix/전체 지정 가능

옵션을 통해 consumer group ID 의 prefix 혹은 전체를 지정 가능할 수 있게 되었다. 기존에는 Spark 에서 지정한 prefix 에 UUID 를 적용해 랜덤하게 생성했다. 보안 팀에서 요구하는 경우가 많다고 한다 (group ID 로 제어하거나 하는 듯).

단, 전체를 지정하는 것은 굉장히 위험하므로 주의해서 사용해야 한다. (Spark 쿼리는 commit offset 관리를 직접 하는데, 토픽 내 모든 레코드를 읽는 것을 보증하기 위해서이다. 이 때 consumer group ID 가 공유되어 실행이 된다면… offset 관리도 꼬이고 레코드도 나누어 읽게 되어 부정확한 결과를 초래하게 된다.)

• [Structured Streaming] Allow users to define prefix of Kafka's consumer group (group.id)
• Allow the user to override the group id of the Kafka's consumer

• Consumer/Producer pool 변화

Consumer pool 의 경우 Apache Commons Pool 이 도입되어 pool 관련 메트릭을 JMX 를 통해 제공하고 기존 pool 구현이 구현 한계로 인해 정상적으로 다루지 못했던 케이스들을 다룰 수 있게 되었다. (예시: self-join)

Apply Apache Commons Pool to KafkaDataConsumer

Producer pool 의 경우 eviction 기준을 재정리하면서 producer 가 idle timeout 보다 길게 사용할 경우 cache 가 오동작하는 경우를 해결했다. (내부적으로는 Guava Cache 를 걷어내고 별도 구현체를 사용했다.)

A cached Kafka producer should not be closed if any task is using it.

다음으로 File Source/Sink 를 짚어 보면…

• 처리가 완료된 소스 파일들을 삭제 혹은 보관 가능

Structured Streaming 쿼리에서 파일 데이터를 처리하는 경우 처리 단위가 파일이기 때문에 필연적으로 수많은 파일들이 처리를 위해 입력 디렉토리에 쌓이게 된다. 기존에는 이미 처리한 파일을 삭제하는 것이 사용자에게 맡겨져 있었고 처리한 파일을 확인하는 작업이 쉽지 않았다. (checkpoint 내 commit log 를 직접 분석해야 하는데 사용자가 파일 포맷까지 분석해서 이걸 해야 될까?)

이제 처리가 완료된 소스 파일들을 Spark 에서 안전하게 삭제하거나 지정한 디렉토리로 이동(보관) 할 수 있게 된다. 별도 쓰레드로 처리가 되어 쿼리 실행 속도에 영향을 크게 주지 않는다.

Delete files after processing in structured streaming

(외 다수)

File Source/Sink 는 DataSource V2 적용으로 재구현되었는데 기존과 다른 변화는 분석해 보지 않아서 잘 모르겠다. Spark+AI summit 2019 에서 다루어졌으니 궁금한 분들은 동영상을 참고하는 것도 방법이겠다.

마지막으로 Source/Sink 외 Structured Streaming 자체 변화를 짚어 보자.

• Structured Streaming UI

DStream 에 한해 제공되던 Streaming 탭이 Structured Streaming 으로 포팅되었다. 기본 UI 를 DStream 에서 가져와서 대부분의 기능이 DStream 에서 제공되던 UI 와 비슷한 것으로 보인다. 마이크로 배치 처리 관련 UI 가 제공되지 않아 streaming query listener 를 통해 정보를 받아서 직접 분석해야 하는 어려움이 있었는데 사용자들에게 큰 도움이 될 것으로 기대가 된다.

Support Structured Streaming UI

• Stream-Stream left outer join 관련 correctness 이슈 수정

Stream-Stream left outer join 사용 시 양 측에서 매칭되어 아웃풋으로 나간 데이터가 특정 조건에서 버그로 인해 한 번도 매칭되지 않은 것으로 오판단되어 null 과 조인되어 한 번 더 아웃풋으로 나가는 버그가 발견되었고, 해당 버그가 수정되었다. 수정을 위해 state 구조의 변화가 필요했고, 그 결과 하위 호환성을 지원하지 못하게 되었다. 기존 2.x 버전의 Stream-Stream join 쿼리를 실행한 checkpoint 를 3.0 에서 읽어들이는 경우 에러가 발생하고 아쉽게도 사용자는 checkpoint 를 버리고 데이터 재처리를 통해 처음부터 다시 실행해야 한다. (Stream-Stream inner join 은 영향을 받지 않는다.)

Stream-stream joins - left outer join gives inconsistent output

• UNION 의 오른쪽 입력으로 stream-stream join 사용 시 crash 이슈 수정

UNION 의 오른쪽 입력으로 stream-stream join 이 사용될 경우, state 의 partition index 가 왼쪽 입력의 partition 수에 영향을 받는 버그가 발견되었다. partition 수는 고정되지 않고 배치 사이에 변경될 수 있는 것이라 정상 동작하던 쿼리가 갑자기 crash 될 가능성이 열려 있는 것이었다. 해당 버그 또한 수정되었는데, 마이그레이션 로직은 아직 제공되지 않았다. 현재로써는 위와 마찬가지로 checkpoint 를 버리고 재처리해야 하는데, 조건 자체가 까다롭기 때문에 아주 많은 사용자가 영향을 받진 않을 것으로 보인다. (실제로 커뮤니티에서 JIRA 이슈가 단 한 번 등록되었고 다른 유저의 댓글도 vote 도 없다)

Failed to get state store in stream-stream join

• Trigger.Once() 실행 시 Data Source 의 limit option 무시

Trigger.Once 는 batch 쿼리를 연속성 있게 실행할 수 있게 하는 특수한 트리거로 기획되었다. 예를 들면 하루에 한 번씩 쿼리를 실행하되 (batch) 데이터를 읽을 위치, 실행 결과 및 상태가 쿼리 실행 간에 연속성 있게 (streaming) 하는 것이 이 트리거의 목적이라고 할 수 있는데, 이런 목적을 제대로 달성하기 위해선 Data Source 의 limit option 이 적용되지 않아야 한다. (일반적으로 Data Source 가 제공하는 limit option 은 개별 micro-batch 가 적당한 양의 데이터를 입력으로 취해 짧게 동작하게 할 수 있게 하는 장치인데 Trigger.Once 는 처리 주기 중간에 누적된 데이터를 모두 처리해야 목적에 부합하게 된다.) 이 부분이 수정되었다.

Introduce AdmissionControl API to Structured Streaming

그 외에도 이벤트 로그 관련 메이저 기능 추가가 있었는데, 주로 스트리밍 쿼리에 도움이 될 변화라 잠깐 소개하자면…

이벤트 로그 파일을 사용자가 지정한 크기로 잘라서 저장할 수 있게 된다. (driver 영역) 그리고 실행 종료된 Job 이나 executor, SQL execution 에 대한 이벤트를 제외하고 다시 기록하는 방식으로 오래된 이벤트 로그 파일들을 축약해서 저장할 수 있게 된다. (History Server 영역)

특히 축약 기능은 스트리밍 어플리케이션 실행 시 꽤 도움이 될 수 있을 것으로 기대하고 있다. 스트리밍 어플리케이션의 특성상 수많은 Job 들이 실행 종료된 상태로 남기 때문에 축약 기능이 로그가 차지하는 공간을 엄청나게 줄일 수 있다.

Allow event logs for running streaming apps to be rolled over.

apache/spark

예시 구현은 mapGroupsWithState API 의 아주 단순한 활용 사례이기 때문에 해당 구현을 이해할 수 있어야 제시하는 구현도 이해할 수 있을 것이다. 해당 코드를 처음 본다면 시간을 내어 코드와 map/flatMapGroupsWithState API 문서를 먼저 읽어보기 바란다.

예시 구현은 프로세싱 타임을 기준으로 세션 윈도우를 정의하고 있다. 프로세싱 타임의 경우에는 늦은 이벤트가 들어오지 않기 때문에 특정 시점에 유효한 세션은 그룹 키당 단 하나만 존재하며, 예시 코드처럼 아주 단순한 타임아웃 처리로도 구현이 가능하다.

이벤트 타임의 경우에는 늦은 이벤트가 들어올 수 있기 때문에 더 많은 경우를 고려해야 한다. 예를 들어 단순히 세션의 간격이 10 초인 세션 윈도우를 상정하고 현재 아래와 같은 세션들이 있다고 가정해 보자. 세션 표현은 (세션 시작 시간, 세션 끝 시간) 으로 하자.

(30, 40) (45, 55) (70, 80)

이벤트 타임의 경우 늦게 도착하는 이벤트 유입이 허용되기 때문에 워터마크의 진행에 따라 여러 세션들이 특정 시점에 동시에 유효할 수 있다.

늦게 도착하는 이벤트들이 허용된다고 하면, 새로운 이벤트가 만들어내는 경우의 수는 프로세싱 타임이 만들어내는 2가지 (기존 세션을 확장, 맨 뒤에 새로운 세션 추가) 가 아니라 총 4가지가 된다.

1. 맨 앞에 새로운 세션 추가
2. 기존 세션을 확장
3. 기존 세션을 확장 & 확장된 세션이 다음 세션과 합쳐져 재확장
4. 맨 뒤에 새로운 세션 추가

하나씩 살펴보자. 1번의 예시는 시간이 15 인 이벤트가 유입되는 것이다. 이벤트가 적용되었을 때 세션들은 아래와 같이 업데이트된다.

(15, 25) (30, 40) (45, 55) (70,80)

2번의 예시는 시간이 32 인 이벤트가 유입되는 것이다. 업데이트된 세션들은 아래와 같다.

(30, 42) (45, 55) (70, 80) // 세션 (30, 40) 와 (32, 42) 가 병합 되어 확장

4번의 예시는 시간이 85 인 이벤트가 유입되는 것이다. 업데이트된 세션들은 아래와 같다.

(30, 40) (45, 55) (70,80) (85, 95)

3번은 1, 2, 4 에 비해서 덜 직관적이고 놓치기 쉽다. 3번의 예시는 시간이 37 인 이벤트가 유입되는 것이다. 업데이트된 세션들은 아래와 같다.

(30, 55) (70, 80) // (30, 40) 과 (37, 47) 이 merge 되어 (30, 47) 로 확장, (30, 47) 과 (45, 55) 가 merge 되어 (30, 55) 로 확장

경우의 수는 살펴보았으니 실제로 state function 을 어떻게 구현해야 할 지 살펴보자. 시작 전, GroupsWithState 클래스 문서 를 먼저 한 번 정독하는 것을 추천한다. (약간 길지만, 주의해야 할 점과 어떻게 동작하는지에 대한 정보가 상세히 적혀 있다.) 문서에 대한 이해가 있으면 왜 이렇게 구현을 했는지에 대한 추측을 쉽게 할 수 있다.

먼저, 세션의 저장용 중간 상태와 출력 상태를 정의해야 한다. 위에서도 언급했지만, 여러 세션들이 동시에 존재할 수 있고 state 에 기록되어야 한다. 즉, 여러 개의 중간 상태들을 저장해야 한다. 최적화된 구현체가 있을 수 있겠지만, 여기서는 단순히 리스트에 기록하기로 하자.

state function 은 크게 두 부분으로 나눌 수 있다. (1) 유입된 데이터를 반영하는 부분, (2) 워터마크가 지나간 유효하지 않은 세션들을 내보내는 부분이다.

코드가 길어 별도 라인 단위 설명은 하지 않고, 적용된 알고리즘과 놓치기 쉬운 부분들에 대한 설명만 하려 한다. 상세한 내용은 링크된 코드를 읽어보기 바란다. (코드와 설명을 같이 읽는 독자들을 위해 코드를 먼저 링크한다. 설명 마지막에 링크가 한 번 더 나온다.)

(1) 부터 살펴보자.

세션 윈도우를 가장 직관적으로 처리하는 방법은 기존 세션 및 이벤트를 정렬한 다음 세션 병합을 적용하는 것이다. 하지만 클래스 문서에는 “이벤트에 대한 정렬은 보장되어 있지 않다" 고 명시되어 있다. 즉, 기존 세션 리스트가 정렬되어 있어도 병합 정렬은 사용할 수 없다. 그러므로 차선으로 기존 세션 리스트는 정렬 상태를 유지하고 삽입 정렬 형태로 이벤트를 기존 세션 리스트에 반영하도록 한다.

반영시 주의할 점이 있다면, 경우의 수 2 번처럼 이벤트가 기존 세션 리스트에 포함되어 세션이 확장되는 경우, 경우의 수 3 번을 다루기 위해 확장된 세션이 앞/뒤 세션과 겹치는지 다시 한 번 확인하고 겹치는 경우 세션을 병합해 줘야 한다는 것이다. 병합 시에 하나의 기존 세션이 삭제되므로 이 부분에 대한 처리도 해 주어야 한다.

모든 이벤트를 반영한 후, 유효하지 않은 세션을 내보내기 위한 타임아웃을 설정해야 한다. 클래스 문서에서도 알 수 있듯이 타임아웃을 설정하지 않으면, watermark 진행으로 인해 유효하지 않은 세션이 발생해도 해당 key 에 대한 이벤트가 유입되지 않으면 세션을 내보낼 수 없다.

여기서 주의해야 할 사항은 state 에 대한 타임아웃은 하나만 설정 가능하다는 것이다. 모든 세션들의 세션 끝 시간에 타임아웃을 설정하는 것이 불가능하다. 우리는 이 기능을 최소한의 트리거 정도로 활용하고, 타임아웃이 트리거되었을 때 유효하지 않은 세션을 직접 찾아서 모두 내보내는 것으로 대응하도록 한다. 타임아웃을 맨 처음 세션의 세션 끝 시간으로 설정한다. 여기서 하나 더 염두에 둘 사항은 watermark 가 타임아웃 시간을 ‘지나가야' 트리거된다는 것이다 .

이제 결과만 정의하면 된다. Append mode / Update mode 에 맞게 적당히 결과를 반환하도록 한다. 여기서는 Update mode 인 경우에만 결과가 발생한다. 필자는 Update mode 에서 실제로 업데이트가 일어난 세션들만 반환하기 위해 로직이 약간 복잡해졌는데, 단순히 특정 키에 업데이트가 일어난 경우 모든 세션을 반환하는 것으로 정의하면 로직이 단순해진다. (의미론 상 세션이 그룹 키에 포함되는 것으로 보아 “그룹키 + 세션” 을 기준으로 변경 사항에 대해 처리했다.)

이제 (2) 에 대해 살펴보자. 위에서도 언급했듯이, 우리는 타임아웃이 트리거되었을 때 유효하지 않은 세션을 직접 찾아서 모두 내보낼 것이다. 세션들이 정렬되어 있다는 점을 활용하면 (유효하지 않은 세션 목록, 유효한 세션 목록) 으로 쉽게 두 부류로 가를 수 있다. 유효한 세션 목록이 존재하지 않는다면, state 를 삭제할 수 있다. 유효한 세션 목록이 존재한다면, state 를 업데이트하고, 타임아웃을 같은 방법으로 다시 적용한다.

그리고 (1) 과 같은 방법으로 결과를 정의한다. 여기서는 Append mode 인 경우에만 결과가 발생한다.

구현된 코드는 아래와 같다. 위의 설명과 비교해 보면서 코드를 읽으면 더 쉽게 이해할 수 있을 것이다.

https://gist.github.com/HeartSaVioR/9a3aeeef0f1d8ee97516743308b14cd6#file-eventtimesessionwindowimplementationviaflatmapgroupswithstate-scala-L32-L189

위의 코드를 이해하면 단순 시간 간격으로 정의된 세션 윈도우 외에도 “세션 끝” 이벤트 등을 반영하거나 이벤트 별로 시간 간격이 다르게 정의되는 세션 윈도우 등을 큰 틀에서 비슷하게 구현할 수 있다.

… 마치며 …

ps. 단순 시간 간격 세션 윈도우의 경우 현재의 타임 윈도우와 동일한 방법으로 사용 가능하도록 하는 패치가 제안되어 있습니다. 위에 링크된 코드에도 SPARK-10816 으로 예시가 포함되어 있습니다. 해당 기능(또는 구현) 에 관심이 있으면 이슈 페이지 SPARK-10816 를 방문해 주시고 VOTE 도 해 주시면 감사하겠습니다. (이상 광고였습니다…!?)

주변에서 Jay Kreps 의 Kafka exactly-once 관련 작성한 글이 돌아다니길래 한 번 훑어 봤다. (영어 실력이 미천해서 정독은 정말 시간을 많이 들여야 되어서… 일단 먼저 훑어봄)

새로운 방법이라도 나온 건가 했는데, 일단 훑어본 걸로는 현재 사용되고 있는 방법을 사용하고 있는 듯 하다. 디테일에 대한 부분은 첫 링크의 Confluent 블로그 글이나 두번째 링크의 KIP 위키 페이지를 보는 것이 더 나아 보인다.

멱등성을 이용하거나 트랜잭션을 이용한 exactly-once 는 Storm trident 때부터 지원한 고전적인 방법이다.

단 한 번 보내는 게 아니라 여러 번 보내되 단 한 번의 시도만 유효하게 만드는 방법이 사용되기 때문에 exactly-once 가 아니라 exactly-once semantic 이라는 주장이 많이 있다. 일반적으로 데이터를 기록하는 연산이 멱등성을 가지고 있거나 트랜잭션을 지원해야 하는 제약사항이 있다.

다른 방법으로 deduplicating 이라고 해서 처리된 메시지의 unique key 를 보관하고 한 번만 처리하는 방법도 사용되고 있다. (Google MillWheel)
https://blog.acolyer.org/…/millwheel-fault-tolerant-stream…/

Storage 로써의 Kafka 는 side-effect 를 고민하지 않아도 되지만, Distributed computation framework 입장에서 보면 state 의 저장만 exactly-once 로 처리되는 것이지 전체 파이프라인이 한 번만 처리되는 게 아니기 때문에 중간 연산에 side-effect 가 없어야 유저가 기대하는 exactly-once 처럼 동작하게 된다.

exactly-once 는 결국 제약사항을 충족해야 달성할 수 있는 건데 정작 홍보할 때는 제약사항은 잘 언급되지 않는다. 그게 반대론자(?) 들이 exactly-once 에 대한 주장을 공격하는 이유 중 하나가 되기도 한다.