Best practice of real-time data analysis based on Confluent+Flink

業務バックグラウンド

実際のビジネス利用においては、リアルタイム PV ・ UV 表示、リアルタイムの売上データ、リアルタイムの店舗 UV、リアルタイムレコメンデーションシステムなど、さまざまなリアルタイムデータ分析が必要です。こうしたニーズに対し、Confluent と Realtime Compute for Apache Flink を組み合わせたソリューションが効果的です。

Confluent は、Apache Kafka の開発元が構築したエンタープライズグレードのフルマネージドストリーミングデータサービスであり、エンタープライズ向け機能によって Kafka の利点を拡張し、Kafka の管理やモニタリングの負担を軽減します。

Realtime Compute for Apache Flink は、Alibaba Cloud が Apache Flink をベースに構築したエンタープライズ向けリアルタイムビッグデータ計算の商用製品です。Apache Flink の創設チームが公式に開発し、グローバル統一の商用ブランドを持ち、包括的な製品マトリックスを提供し、オープンソースの Flink API と完全互換であり、強力な Alibaba Cloud プラットフォームをベースにクラウドベースの商用付加価値機能を提供します。

1. 準備 - Confluent クラスターと Realtime Compute for Apache Flink クラスターの作成

1. Confluent 管理コンソールにログインし、Confluent クラスターを作成します。作成手順については、Confluent クラスターの有効化ガイドをご参照ください。

2. Realtime Compute for Apache Flink 管理コンソールにログインし、VVP クラスターを作成します。VVP クラスター作成時に選択する VPC は、Confluent クラスターと同じリージョンおよび同じ VPC である必要があります。これにより、VVP 内から Confluent の内部ドメイン名にアクセスできます。

2. ベストプラクティス - プレイヤー課金額のリアルタイム統計 - Confluent + Realtime Compute for Apache Flink + Hologres

2.1 Confluent メッセージキューの新規作成

1. Confluent クラスターリストページからコントロールセンターにログインします。

2. 左側の [Topics] を選択し、[Add a topic] をクリックして、confluent-vvp-test という名前の Topic を作成し、パーティションを 3 に設定します。

2.2 Hologres 結果テーブルの設定

1. Hologres コンソールにアクセスし、Hologres インスタンスを選択し、DB 管理でデータベース mydb を追加します。

2. Hologres データベースにログインし、新規 SQL を作成します。

3. Hologres で結果テーブルの SQL ステートメントを作成します。

2.3 Realtime Compute for Apache Flink VVP ジョブの作成

1. VVP コンソールにログインし、クラスターが存在するリージョンを選択し、コンソールをクリックして開発インターフェイスに入ります。

2. [ジョブ開発] タブをクリックし、[新規ファイル] をクリックします。ファイル名は confluent-vvp-hologres、ファイルタイプはストリームジョブ / SQL です。

3. 入力ボックスにコードを記述します。

4. 詳細設定で、依存ファイル truststore.jks を追加します (内部ドメイン名にアクセスする場合はこのファイルが必要ですが、パブリックドメイン名へのアクセスには不要です)。依存ファイルへの固定パスプレフィックスは /link/usrlib/ です (実際のパスは /link/usrlib/truststore.jks)。

5. [オンライン] ボタンをクリックしてオンライン化を完了します。

6. 運用保守ロールリストで新規オンライン化したロールを見つけ、[開始] ボタンをクリックし、ステータスが [実行中] に更新されるまで待ちます。

7. コントロールセンターの [Topics -> Messages] ページで、テストメッセージを 1 つずつ送信します。

2.4 ユーザー課金額のリアルタイム統計効果の確認

3. ベストプラクティス - E コマースのリアルタイム PV ・ UV 統計 - Confluent + Realtime Compute for Apache Flink + ApsaraDB RDS

3.1 Confluent メッセージキューの新規作成

1. Confluent クラスターリストページからコントロールセンターにログインします。

2. 左側の [Topics] を選択し、[Add a topic] をクリックして、pv-uv という名前の Topic を作成し、パーティションを 3 に設定します。

3.2 ApsaraDB RDS 結果テーブルの作成

1. ApsaraDB RDS 管理コンソールにログインし、ApsaraDB RDS を購入します。ApsaraDB RDS と Realtime Compute for Apache Flink フルマネージドクラスターが同じリージョンで同じ VPC 配下にあることを確認してください。

2. 仮想スイッチネットワークセグメント (vSwitch IP セグメント) を ApsaraDB RDS ホワイトリストに追加します。詳細はホワイトリスト設定ドキュメントをご参照ください。

3. [vSwitch IP セグメント] は Flink ワークスペースの詳細で確認できます。

4. [アカウント管理] ページでアカウント [特権アカウント] を作成します。

5. データベースインスタンスの [データベース管理] で新規データベース [confluent_vvp] を作成します。

6. システムが提供する DMS (Data Management) サービスを使用して ApsaraDB RDS にログインし、上記で作成した特権アカウントのログイン名とパスワードを入力します。

7. [confluent_vvp] データベースをダブルクリックして SQLConsole を開き、次のテーブル作成ステートメントを SQLConsole にコピーアンドペーストして、結果テーブルを作成します。

CREATE TABLE result_ cps_ total_ summary_ pvuv_ min(

summary_ Date date NOT NULL COMMENT 'Statistics date',

summary_ Min varchar (255) COMMENT 'Count minutes',

pv bigint COMMENT 'pv',

uv bigint COMMENT 'uv',

Currenttime timestamp COMMENT 'Current time',

primary key(summary_date,summary_min)

)

3.3 Realtime Compute for Apache Flink VVP ジョブの作成

1. [VVP コンソール] 新規ファイルを作成します。

2. SQL フィールドにコードを入力します。

3. [オンライン] をクリックした後、運用保守ページで [開始] ボタンをクリックし、ステータスが RUNNING に更新されるまで待ちます。

4. コントロールセンターの [Topics -> Messages] ページで、テストメッセージを 1 つずつ送信します。

3.4 PV と UV の効果の確認

ApsaraDB RDS データテーブルの PV と UV が、送信されたメッセージデータに応じて動的に変化することが確認できます。同時に、[データビジュアライゼーション] で対応するチャート情報を確認できます。