メインコンテンツへ移動

Flink のアップグレード（クロスバージョン）

  1. ConfigMap を削除して Flink の Job 一覧を消去します

    以下のスクリプトで ConfigMap を一括削除します。

    configmap "md-flink" deleted などのメッセージが表示されない場合、名前空間が誤っているか、ConfigMap が存在しません。再確認してください。

    # テンプレート：<namespace> を Flink をデプロイした実際の名前空間に置き換えてください
    for i in $(kubectl -n <namespace> get cm | awk '$1~"md-flink"{print $1}'); do kubectl -n <namespace> delete cm $i; done

    # 例（名前空間が flink の場合）：
    for i in $(kubectl -n flink get cm | awk '$1~"md-flink"{print $1}'); do kubectl -n flink delete cm $i; done

  2. すべての Flink ノードに、アーキテクチャに対応するイメージをインポートします（オフラインパッケージダウンロード）。

    crictl pull nocoly/flink:バージョン番号

  3. 設定ファイルを修正します

    flink.yaml内で、flink-jobmanager及びflink-taskmanagerサービスで利用するイメージバージョンを修正します。

    - name: jobmanager
      image: nocoly/flink:<version>
    - name: taskmanager
      image: nocoly/flink:<version>
    v1.19.710 へのアップグレード時に必要な flink.yaml の追加変更を表示

    1. metrics で始まる既存の設定をすべて削除し、以下の Kafka メトリクス送信設定を追加してください（Kafka アドレスは実際の環境に合わせて変更）。

       metrics.job.status.enable: STATE
       metrics.reporters: kafka_reporter,kafka_reporter_running,kafka_reporter2,kafka_reporter_running2
       metrics.reporter.kafka_reporter.factory.class: org.apache.flink.metrics.kafka.KafkaReporterFactory
       metrics.reporter.kafka_reporter.bootstrap.servers: 192.168.10.7:9092,192.168.10.8:9092,192.168.10.9:9092 # Kafka アドレス
       metrics.reporter.kafka_reporter.chunk.size: 20000
       metrics.reporter.kafka_reporter.interval: 60s
       metrics.reporter.kafka_reporter.filter.metrics: numRecordsIn,numRecordsOut,runningTime
       metrics.reporter.kafka_reporter.topic: flink_metrics_counter
       metrics.reporter.kafka_reporter.taskNamePrefix: HDP0x5c2_
       metrics.reporter.kafka_reporter_running.factory.class: org.apache.flink.metrics.kafka.KafkaReporterFactory
       metrics.reporter.kafka_reporter_running.bootstrap.servers: 192.168.10.7:9092,192.168.10.8:9092,192.168.10.9:9092 # Kafka アドレス
       metrics.reporter.kafka_reporter_running.chunk.size: 20000
       metrics.reporter.kafka_reporter_running.interval: 60s
       metrics.reporter.kafka_reporter_running.filter.metrics: RUNNINGState
       metrics.reporter.kafka_reporter_running.topic: flink_metrics_gauge
       metrics.reporter.kafka_reporter_running.taskNamePrefix: HDP0x5c2_
       metrics.reporter.kafka_reporter2.factory.class: org.apache.flink.metrics.kafka.KafkaReporterFactory
       metrics.reporter.kafka_reporter2.bootstrap.servers: 192.168.10.7:9092,192.168.10.8:9092,192.168.10.9:9092 # Kafka アドレス
       metrics.reporter.kafka_reporter2.chunk.size: 20000
       metrics.reporter.kafka_reporter2.interval: 60s
       metrics.reporter.kafka_reporter2.filter.metrics: numRecordsIn,numRecordsOut,runningTime
       metrics.reporter.kafka_reporter2.topic: flink_metrics_counter-hdp
       metrics.reporter.kafka_reporter2.taskNamePrefix: HDP0x5c2_
       metrics.reporter.kafka_reporter_running2.factory.class: org.apache.flink.metrics.kafka.KafkaReporterFactory
       metrics.reporter.kafka_reporter_running2.bootstrap.servers: 192.168.10.7:9092,192.168.10.8:9092,192.168.10.9:9092 # Kafka アドレス
       metrics.reporter.kafka_reporter_running2.chunk.size: 20000
       metrics.reporter.kafka_reporter_running2.interval: 60s
       metrics.reporter.kafka_reporter_running2.filter.metrics: RUNNINGState
       metrics.reporter.kafka_reporter_running2.topic: flink_metrics_gauge-hdp
       metrics.reporter.kafka_reporter_running2.taskNamePrefix: HDP0x5c2_

    2. kind: Role の設定で、rules.verbs に configmaps リソースの patch 権限を追加します。

      kind: Role
      apiVersion: rbac.authorization.k8s.io/v1
      metadata:
        name: configmap-access
        namespace: default
      rules:
      - apiGroups: [""]
        resources: ["configmaps"]
        verbs: ["update", "get", "watch", "list", "create", "edit", "delete", "patch"] # patch 権限を追加

  4. サービスを再起動します

    kubectl apply -f flink.yaml

  5. アップグレード完了後、同期タスク一覧で対象タスクを手動で再度有効化／公開してください。これにより同期タスクが正常に動作するようになります。