Demo-ストリーム処理の消費
ダウンロード
フォーカスモード
フォントサイズ
本文では、Tencent Cloud Oceanus、Flinkという2つのストリーム処理フレームワークを使用して、ログをカスタム消費する方法を紹介します。
Tencent Cloud Oceanus
1. Oceanusコンソールを開き、依存関係を新規作成し、flink-connector-cls-1.0.0.jarパッケージをアップロードします。
2. SQLジョブを新規作成します。Flinkバージョンは Flink-1.16を選択してください。
3. ジョブパラメータを設定し、手順1でアップロードしたjarパッケージの依存関係を選択して確認します。その他のジョブパラメータは必要に応じて調整できます。
4.
テーブル作成文を編集します。
CREATE TABLE `cls_source_table` (`key1` STRING,`key2` STRING,`key3` STRING,`key4` STRING,`__TIMESTAMP__` BIGINT) WITH ('connector' = 'cls','region' = '地域、例：ap-guangzhou'、'logset_id' = 'ログセットID、1つのログセットのみサポートされます'、'topic_ids' = 'ログトピックID。複数のトピックを使用する場合は、「,」で区切ってください'、'access_id' = 'お客様のSecret_id。Tencent Cloud CAMで取得してください'、'access_key' = 'お客様のSecret_key。Tencent Cloud CAMで取得してください'、'consumer_group_name' = 'コンシューマーグループ名。例：flink-connector-cls-5bXXXXb2'、'offset_start_time' = 'begin','internal' = 'true','format' = 'json','scan.parallelism' = '3');
パラメータの説明は以下の通りです：
テーブルフィールドはCLS側のログフィールドと一致する必要があります。そのうち、__TIMESTAMP__はログ時間であり、必要に応じて使用できます。その他のメタデータフィールド（__FILENAME__、__SOURCE__、__TAG__など）はまだ追加されていません。
connectorは必須項目であり、clsである必要があります。
詳細については、カスタムコンシューマーパラメータの説明を参照してください。
注意：
現在、connector自体はチェックポイントに基づくジョブリカバリをサポートしています。停止するたびにスナップショット作成プロセスをトリガーする必要があります。
スナップショットから復旧した後、checkpointのポイントに基づいて引き続き消費でき、Exactly Onceセマンティクスを保証します。異常障害でダウンした場合、checkpointが正常にトリガーされず、以前のcheckpointに基づいて復旧する際にデータが重複する可能性があり、At Least Onceセマンティクスのみを保証します。
セルフホスト型Flink
1. Java 8バージョンのOpenJDKをインストールしてください。（CentOS/RHELシステム）
1.1 競合を避けるため、システムにJavaがインストールされているかどうかを先に確認してください。
java -version
未インストールの場合は、以下のコマンドを実行してOpenJDK 8をインストールしてください。
sudo yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
1.2 インストール結果を確認します（重要な手順です。バージョンが正しいことを確認してください）。
java -versionjavac -version # javacも正常にインストールされていることを確認してください（develパッケージにはコンパイラが含まれており、Flinkに必要です）
2. Flink-1.16.3バージョンを例として、Flinkクラスタをダウンロードしてインストールして起動してください。
2.1 Flinkをダウンロードし、権限を設定してください。
# 1. Flink 1.16.3（scala 2.12バージョン、主流バージョン）をダウンロードします。wget https://archive.apache.org/dist/flink/flink-1.16.3/flink-1.16.3-bin-scala_2.12.tgz# 2. 指定ディレクトリに解凍します（推奨：/usr/local/）sudo tar zxvf flink-1.16.3-bin-scala_2.12.tgz -C /usr/local/# 3. ディレクトリの名前を変更します（後続の操作を簡素化するため）sudo mv /usr/local/flink-1.16.3 /usr/local/flink# 4. 一般ユーザーに操作権限を付与します（後続の起動権限の問題を回避するため）sudo chown -R $USER:$USER /usr/local/flink
2.2 Flinkクラスタを起動してください。
# 1. Flinkインストールディレクトリに移動します。cd /usr/local/flink# 2. クラスタを起動します（シングルノード：1つのJobManager + 1つのTaskManager）./bin/start-cluster.sh
2.3 起動状態を確認してください（重要な手順です）。
方式1：起動後、JDK付属のjpsコマンドでJavaプロセスを確認できます。TaskManagerプロセスが起動しているかどうかを確認できます。
jps
正常な出力には以下が含まれます：
- StandaloneSessionClusterEntrypoint（JobManagerプロセス）- TaskManagerRunner（TaskManagerプロセス）- Jps（自身のプロセス）
方式2：Web UIにアクセスします（クラスタにアクセスできることを確認します）。
ブラウザでアクセスします：
http://サーバーIP:8081。Flinkコンソールが表示され、Task Managersの数が1であれば、クラスタの起動は成功です。3. SQL clientを起動してテーブルを作成します。Flink SQLコマンドラインに入り、テーブル作成文を編集します。
3.1 Flinkインストールディレクトリに移動します（パスが正しいことを確認してください）。
cd /usr/local/flink
3.2 SQL Client（対話式コマンドラインツール）を起動します。
./bin/sql-client.sh
起動成功後、以下の内容が表示されます：
4. Flinkクラスタを終了します。
./bin/stop-cluster.sh
5. Flink Web UIにアクセスしてタスクの実行状況を確認します：
http://localhost:8081/#/overview。フィードバック