クラウドネイティブに基づくFlinkコンピューティングプラットフォームの実践

著者: 王傑、所属: 中国移動スマートホームオペレーションセンター

ラボガイド

クラウドネイティブの普及に伴い、ますます多くのバックエンドアプリケーションがコンテナーに移行され、k8s を通じてオーケストレーションおよび管理されるようになりました。近年では、FlinkやSparkなどビッグデータ分野のコンピューティングエンジンもk8sへのサポートを表明しており、ビッグデータアプリケーションは従来のyarn時代からクラウドネイティブ時代へと移行しています。この記事では、Flink と k8s を主な技術手段として使用し、クラウドネイティブコンピューティングプラットフォームの構築方法を紹介します。

パート01 K8sの概要

Kubernetes (k8s または kube とも呼ばれる) は、Linux コンテナの操作を自動化するオープンソースプラットフォームです。これにより、ユーザーはアプリケーションのコンテナ化プロセスにおける多くの手動の展開および拡張操作を節約できます。つまり、Linux コンテナを実行しているホストのグループを集約できます。 K8s を使用すると、これらのクラスターを簡単かつ効率的に管理できます。さらに、これらのクラスターは、パブリッククラウド、プライベートクラウド、またはハイブリッドクラウドのホスト全体に展開できます。アーキテクチャ図は次のとおりです。

注記：

kubectl: Kubernetes API を使用して Kubernetes クラスターのコントロールプレーンと通信するためのコマンドラインツール。
ETCD: 分散システムまたはコンピュータクラスターの構成、サービス検出、およびスケジュール調整に使用される分散キー値ストレージ。
ノード: リクエストと割り当てられたタスクの実行を担当するコンピューター。
ポッド (コンテナセット): 単一のノードにデプロイされ、1 つ以上のコンテナを含むコンテナグループ。
Kubelet: コンテナマニフェストを読み取り、指定されたコンテナが稼働していることを確認する、ノード上で実行されるサービス。

1. マスター上には、コントローラー、API サーバー、スケジューラー、ストレージ用の Etcd などから構成されます。 etcd はマスターとして扱うことも、マスターから独立したストレージとして扱うこともできます。マスターのコントローラー、API サーバー、およびスケジューラはすべて個別のプロセスモードです。これは、マスター全体が単一プロセスモデルである Yarn とは多少異なります。 K8s マスターは複数のマスター間で自発的な選挙を完了することもでき、アクティブ状態のマスターが外部にサービスを提供します。

2. スレーブには、主に Kube プロキシ、Kubelet、Docker などの関連コンポーネントが含まれます。各ノードにデプロイされる関連コンポーネントは類似しており、ノード上で実行されている複数のポッドを管理するために使用されます。

3. さまざまなユーザーの習慣に応じて、 UI または CLI を介してタスクを K8s に送信できます。ユーザーは、K8s が提供するダッシュボード Web UI または Kubectl コマンドラインを通じてタスクを送信できます。

パート02 Flinkの概要

Apache Flink は、無制限および制限付きデータストリームでのステートフル計算のためのフレームワークおよび分散処理エンジンです。 Flink はすべての一般的なクラスター環境で実行され、メモリ速度とあらゆる規模で計算を実行できます。アーキテクチャ図は次のとおりです。

Flink システム全体は 3 つの部分で構成されています。

クライアント: Flink クライアントは主に、ユーザータスク (ストリーミングジョブ) を Flink システムに送信する機能をユーザーに提供します。

TaskManager:特定のユーザータスクを実行する Flink システムのビジネス実行ノード。複数の TaskManager が存在する可能性があり、すべての TaskManager は同等です。

JobManager: Flink システムの管理ノードはすべての TaskManager を管理し、どの TaskManager でユーザータスクを実行するかを決定します。 HA モードでは複数の JobManager が存在できますが、マスター JobManager は 1 つだけです。

パート 03 K8s デプロイメントモードの Flink

Flink は、ローカル、スタンドアロン、Yarn、Docker、Kubernetes モードなど、さまざまな方法でデプロイできます。 Flink ジョブの送信モードに応じて、次の 3 つのモードに分けられます。

モード1: アプリケーションモード

Flink によって送信されたプログラムはクラスターの内部アプリケーションとして扱われ、クライアントは面倒な準備作業を行う必要がなくなります。 (たとえば、メイン関数の実行、JobGraph の生成、依存関係のダウンロードと各ノードへの配布など) メイン関数は実行のために JobManager に送信されます。 1 つのアプリケーション、1 つのクラスターインスタンス。

モード 2: ジョブごとのモード

これは、より強力なリソース分離を備えた Yarn、Mesos などのリソース管理フレームワークを使用するクライアントモードのアプリケーションモードとして理解できます。 1 つのジョブ、1 つのクラスターインスタンス。だんだんと放置されていった。

モード3: セッションモード

これは、準備作業を行うためにクライアントによって提出されることもあります。ただし、クラスターインスタンスは作成されており、すべてのジョブによって共有されます。 1 つのジョブによって発生した JobManager の障害により、すべてのジョブが失敗する可能性があります。

一般に、k8s ではスタンドアロンセッション、スタンドアロンアプリケーション、ネイティブセッション、ネイティブアプリケーションの 4 つのモードを実行できます。その特徴は次のとおりです。

上記の特性に基づいて、ネイティブアプリケーションモードは一般に本番環境への展開に使用されます。起動フローチャートは次のとおりです。

1. サービス、マスター、および ConfigMap リソースを作成すると、 Flink マスターデプロイメントにはすでにユーザー Jar が含まれます。この時点で、クラスターエントリポイントはユーザー Jar からユーザーのメインを抽出または実行し、JobGraph を生成します。次に、Dispatcher に送信され、Dispatcher はマスターを生成し、ResourceManager からリソースを適用します。以降のロジックはセッションの場合と同じです。

2. セッションとの最大の違いは、1 つのステップで送信されることです。 2 段階の送信は必要ないため、タスク開始後に外部 UI にアクセスする必要がない場合は、外部サービスは必要ありません。タスクを 1 ステップで直接送信して実行できます。 K8s ApiServer を使用したローカルポート転送またはプロキシを介して Flink の Web UI にアクセスできます。この時点で、外部サービスは必要なくなります。つまり、LoadBalancer または NodePort を占有する必要はありません。

パート 04 ネイティブ k8s デプロイメントの実践における Flink

Flink をデプロイする前に、次の要件を満たす実行中の k8s クラスターが必要です。

Kubernetes >= 1.9。
Pod と Service を一覧表示、作成、削除できる KubeConfig は、~/.kube/config を通じて設定できます。 kubectl auth can-i <list|create|edit|delete> pods を実行して権限を確認できます。
Kubernetes DNS を有効にします。
Pod を作成および削除するための RBAC 権限を持つデフォルトのサービスアカウント。

上記の k8s クラスターを作成したら、次の手順を実行して k8s ネイティブアプリケーションタスクを開始する必要があります。

1. flink-clusterという名前の名前空間を作成する

 kubectl 名前空間 flink-cluster を作成します

2. アカウントを作成する

 Kubectl create serviceaccount flink -n flink-cluster

3. サービスアカウントとロールバインディング

 kubectl create clusterrolebinding flink-role-binding-flink \
 --clusterrole=編集\
 --serviceaccount=flink-cluster:flink

4. Dockerfileを書く

 # ベースイメージ
apache/flink:1.14.5-scala_2.11 より
mkdir -p $FLINK_HOME/usrlibを実行します。
 mkdir -p $FLINK_HOME/hadoopconfを実行します。
 flink-1.14.5/examples/streaming/TopSpeedWindowing.jar $FLINK_HOME/usrlib/TopSpeedWindowing.jar をコピーします。
 core-site.xml をコピー $FLINK_HOME/hadoopconf
 hdfs-site.xml $FLINK_HOME/hadoopconf をコピーします
# hdfs を読み込むために hdfs 関連の jar を追加します
flink-shaded-hadoop2-uber-2.8.3-1.8.3.jar を $FLINK_HOME/lib にコピーします。
 # チェックポイントは hdfs に保存されており、hdfs に読み書きする必要があるため、hadoop の core-site.xml と hdfs-site.xml を取得するように HADOOP_CONF_DIR を設定します。
環境変数 HADOOP_CONF_DIR=$FLINK_HOME/hadoopconf:$HADOOP_CONF_DIR

5. 鏡

 Apache/flink:v0.1 を実行します。

ミラーディレクトリに存在するファイル

コアサイト.xml
 Dockerファイル
フリンク-1.14.5
 flink-shaded-hadoop2-uber-2.8.3-1.8.3.jar
 hdfs-サイト.xml

6. アプリケーションモードタスクを開始する

 ./flink-1.14.5/bin/flink 実行アプリケーション \
 --target kubernetes-application \
 -Dkubernetes.namespace=flink-cluster \
 -Dkubernetes.jobmanager.service-account=flink \
 -Dkubernetes.cluster-id=最初のアプリケーション クラスター \
 -Dkubernetes.high-availability=org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory \
 -Dkubernetes.cluster-id=最初のアプリケーション クラスター \
 -Dhigh-availability.storageDir=hdfs://${HDFS クラスター名}/flink/recovery \
 -Dkubernetes.container.image=apache/flink:v0.1 \
 -Dkubernetes.rest-service.exposed.type=NodePort \
 -Dstate.backend=rocksdb \
 -Dstate.checkpoints.dir=hdfs://${HDFS クラスター名}/flink/flink-checkpoints \
 -Dstate.backend.incremental=true \
 ローカル:///opt/flink/usrlib/TopSpeedWindowing.jar

起動後、最後に以下のログが表示されます。最後のアドレスは、タスクの Web UI アドレスです。

 2022-07-27 16:45:00,320 INFO org.apache.flink.kubernetes.KubernetesClusterDescriptor [] - flink アプリケーション クラスター my-first-application-cluster が正常に作成されました。JobManager Web インターフェース: http://ip:port

起動後、k8s が次の flink コンポーネントを開始したことがわかります。

 [root@master1 ~]# kubectl get svc -n flink-cluster
名前 タイプ クラスター IP 外部 IP ポート 年齢
my-first-application-cluster ClusterIP なし <なし> 6123/TCP、6124/TCP 1h
 my-first-application-cluster-rest ClusterIP ip xxxx <なし> 8082/TCP 46h
 [root@master1 ~]# kubectl get デプロイメント -n flink-cluster
名前 準備完了 最新 利用可能 年齢
私の最初のアプリケーション クラスター 1/1 1 1 1h
 [root@master1 ~]# kubectl get pods -n flink-cluster
名前 準備完了 ステータス 再起動 年齢
my-first-application-cluster-7c4d9d7994-6vwjr 1/1 実行中 0 1h
 my-first-application-cluster-taskmanager-1-1 1/1 実行中 0 461h

これは、Flink が正常に起動したことを意味します。 k8s には jobmanger と taskmanger という 2 つのポッドと、ポートマッピング用のサービスがあることがわかります。

パート05 Flinkプラットフォームk8s変換の実践

k8s 変換の前に、すでに Flink コンピューティングプラットフォームがありました。主に 3 つの部分で構成されます。Web UI はユーザーが SQL タスクまたは JAR タスクを送信するために使用され、Web サーバーは SQL または JAR を解析するために使用され、次に Flink クライアント送信コマンドを生成し、最後にコマンドを実行してクラスターにタスクを送信します。

上の図に示すように、当社のプラットフォームにはすでに Yarn モードとスタンドアロンモードのサブミッションがあります。クラウドネイティブをサポートするには、元のベースにネイティブ k8s 送信メソッドを追加するだけです。コードレベルでは、タスク送信抽象クラスにネイティブ k8s 実装を追加します。

次の図は、特定のネイティブ k8s SQL タスク送信ページを示しています。 Flink タスクの送信を完了するには、実行モード、実行パラメータ、および依存するサードパーティの jar パッケージを設定し、特定の SQL スクリプトを追加するだけです。