如何使用Flink runner将Beam Python任务提交到Kubernetes上？

我想在Kubernetes内的Flink runner上使用Beam运行一个连续的流处理作业。我一直在这里学习这个教程（https://python.plainenglish.io/apache-beam-flink-cluster-kubernetes-python-a1965f37b7cb），但我不确定作者在谈论“flink主容器”时指的是什么。我不明白我应该如何将我的Python代码提交到集群中，当代码在容器映像本身中定义时。
Kubernetes Flink集群架构看起来像这样：

单个JobManager，通过Service和Ingress暴露Flink Web UI
*多个任务管理器，每个任务管理器运行2个容器：
Flink任务管理器
Beam worker池，暴露端口50000

示例教程中的Python代码具有Beam配置，如下所示：

options = PipelineOptions([
    "--runner=FlinkRunner",
    "--flink_version=1.10",
    "--flink_master=localhost:8081",
    "--environment_type=EXTERNAL",
    "--environment_config=localhost:50000"
])

很明显，当您按照教程在本地运行它时，它会与Beam worker池对话以启动应用程序。但是，如果我有一个包含我的应用程序代码的Docker镜像，并且我想在Kubernetes中启动这个应用程序，我应该在Kubernetes集群中的哪里部署这个镜像？是否作为eachTask Manager pod中的容器（因此使用localhost：50000与Beam通信）？或者我创建一个包含我的应用程序代码的单个pod，并将该pod指向我的任务管理器的端口50000-如果是这样，我有多个任务管理器是一个问题吗？
任何指向文档或示例的指针都将非常有帮助。这个other SO question有一个不完整的答案。

使用flink的作业管理器和任务管理器的自定义图像
Dockerfile

FROM apache/flink:1.16.2-scala_2.12-java11
ARG FLINK_VERSION=1.16.2
ARG KAFKA_VERSION=2.8.0

# Install python3.7 
RUN set -ex; \
  apt-get update && \
  apt-get install -y build-essential libssl-dev zlib1g-dev libbz2-dev libffi-dev lzma liblzma-dev && \
  wget https://www.python.org/ftp/python/3.7.9/Python-3.7.9.tgz && \
  tar -xvf Python-3.7.9.tgz && \
  cd Python-3.7.9 && \
  ./configure --without-tests --enable-shared && \
  make -j4 && \
  make install && \
  ldconfig /usr/local/lib && \
  cd .. && rm -f Python-3.7.9.tgz && rm -rf Python-3.7.9 && \
  ln -s /usr/local/bin/python3 /usr/local/bin/python && \
  ln -s /usr/local/bin/pip3 /usr/local/bin/pip && \
  apt-get clean && \
  rm -rf /var/lib/apt/lists/* && \
  python -m pip install --upgrade pip; \
  pip install apache-flink==${FLINK_VERSION}; \
  pip install kafka-python

RUN pip install --no-cache-dir apache-beam[gcp]

# Copy files from official SDK image, including script/dependencies.
COPY --from=apache/beam_python3.7_sdk:2.48.0 /opt/apache/beam/ /opt/apache/beam/

此外，在上述部署中，没有卷来存储staging-artifacts。因此，像这样创建PVC，并根据storageclassName调整PVC

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: staging-artifacts-claim
  namespace: flink
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 5Gi
  storageClassName: standard

使用flink的自定义映像的部署yaml应该是这样的

apiVersion: v1
kind: Service
metadata:
  name: flink-jobmanager
  namespace: flink
spec:
  type: ClusterIP
  ports:
  - name: rpc
    port: 6123
  - name: blob-server
    port: 6124
  - name: webui
    port: 8081
  selector:
    app: flink
    component: jobmanager
---
apiVersion: v1
kind: Service
metadata:
  name: beam-worker-pool
  namespace: flink
spec:
  selector:
    app: flink
    component: taskmanager
  ports:
    - protocol: TCP
      port: 50000
      targetPort: 50000
      name: pool
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-jobmanager
  namespace: flink
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink
      component: jobmanager
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      containers:
      - name: jobmanager
        image: custom-flink:latest
        imagePullPolicy: IfNotPresent
        args: ["jobmanager"]
        ports:
        - containerPort: 6123
          name: rpc
        - containerPort: 6124
          name: blob-server
        - containerPort: 8081
          name: webui
        livenessProbe:
          tcpSocket:
            port: 6123
          initialDelaySeconds: 30
          periodSeconds: 60
        volumeMounts:
        - name: flink-config-volume
          mountPath: /opt/flink/conf
        - name: flink-staging
          mountPath: /tmp/beam-artifact-staging 
        securityContext:
          runAsUser: 9999
      volumes:
      - name: flink-config-volume
        configMap:
          name: flink-config
          items:
          - key: flink-conf.yaml
            path: flink-conf.yaml
          - key: log4j-console.properties
            path: log4j-console.properties
      - name: flink-staging
        persistentVolumeClaim:
          claimName:  staging-artifacts-claim 
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-taskmanager
  namespace: flink
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink
      component: taskmanager
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
      - name: taskmanager
        image: custom-flink:latest
        imagePullPolicy: IfNotPresent
        args: ["taskmanager"]
        ports:
        - containerPort: 6122
          name: rpc
        - containerPort: 6125
          name: query-state
        livenessProbe:
          tcpSocket:
            port: 6122
          initialDelaySeconds: 30
          periodSeconds: 60
        volumeMounts:
        - name: flink-config-volume
          mountPath: /opt/flink/conf/
        - name: flink-staging
          mountPath: /tmp/beam-artifact-staging 
          runAsUser: 9999
      - name: beam-worker-pool
        image: apache/beam_python3.11_sdk
        args: ["--worker_pool"]
        ports:
        - containerPort: 50000
          name: pool
        livenessProbe:
          tcpSocket:
            port: 50000
          initialDelaySeconds: 30
          periodSeconds: 60
        volumeMounts:
        - name: flink-staging
          mountPath: /tmp/beam-artifact-staging 
      - name: flink-config-volume
        configMap:
          name: flink-config
          items:
          - key: flink-conf.yaml
            path: flink-conf.yaml
          - key: log4j-console.properties
            path: log4j-console.properties
      - name: flink-staging 
        persistentVolumeClaim:
          claimName:  staging-artifacts-claim

以下是Flinkpipeline选项

flink_options = PipelineOptions([
        "--runner=FlinkRunner",
        "--flink_master=flink-jobmanager:8081",
        "--environment_type=EXTERNAL",
        "--environment_config=beam-worker-pool:50000",
    ])

从另一个容器运行Beam代码可以是cronjob/job或deployment，要使用python sdk运行Beam代码，请使用python代码创建一个映像，并确保使用apache/beam_python3.7_sdk:2.48.0并在其中安装java，以便扩展服务将启动，否则它将使用docker。
请根据需要调整sdk的版本

如何使用Flink runner将Beam Python任务提交到Kubernetes上？

2条答案

相关问题

热门标签

最新问答