Pub/Sub to Redis テンプレート

Pub/Sub to Redis テンプレートは、Pub/Sub サブスクリプションからメッセージを読み取り、メッセージ ペイロードを Redis に書き込むストリーミング パイプラインです。このテンプレートの最も一般的なユースケースは、ログを Redis Enterprise にエクスポートして、高度な検索ベースのログ分析をリアルタイムで行うことです。

  • Redis に書き込む前に、JavaScript ユーザー定義関数をメッセージ ペイロードに適用できます。
  • 処理が失敗したメッセージは、トラブルシューティングと再処理のために Pub/Sub の未処理のトピックに転送されます。
  • セキュリティを強化するため、データベース エンドポイント接続を設定するときに SSL 接続を有効にします。このテンプレートは相互 TLS をサポートしていません。

パイプラインの要件

  • パイプラインの実行前にソース Pub/Sub サブスクリプションが存在している必要があります。
  • パイプラインを実行する前に、Pub/Sub に未処理のトピックが存在している必要があります。
  • Redis データベース エンドポイントは、Dataflow ワーカーのサブネットワークからアクセスできる必要があります。

テンプレートのパラメータ

必須パラメータ

  • inputSubscription: 入力の読み取り元の Pub/Sub サブスクリプション。形式は projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_ID> です(例: projects/your-project-id/subscriptions/your-subscription-name)。
  • redisHost: Redis データベース ホスト(例: your.cloud.db.redislabs.com)。デフォルトは 127.0.0.1 です。
  • redisPort: Redis データベース ポート(例: 12345)。デフォルトは 6379 です。
  • redisPassword: Redis データベースのパスワード。デフォルトは空です。

オプション パラメータ

  • sslEnabled: Redis データベースの SSL パラメータ。デフォルトは false です。
  • redisSinkType: Redis シンク。サポートされている値は STRING_SINK, HASH_SINK, STREAMS_SINK, and LOGGING_SINK です(例: STRING_SINK)。デフォルトは STRING_SINK です。
  • connectionTimeout: Redis 接続タイムアウト(ミリ秒)(例: 2000)。デフォルトは 2000 です。
  • ttl: の有効期限(秒)。HASH_SINKttl のデフォルトは -1 で、期限切れになることはありません。
  • javascriptTextTransformGcsPath: 使用する JavaScript ユーザー定義関数(UDF)を定義する .js ファイルの Cloud Storage URI(例: gs://my-bucket/my-udfs/my_file.js)。
  • javascriptTextTransformFunctionName: 使用する JavaScript ユーザー定義関数(UDF)の名前。たとえば、JavaScript 関数コードが myTransform(inJson) { /*...do stuff...*/ } の場合、関数名は myTransform です。JavaScript UDF の例については、UDF の例(https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples)をご覧ください。
  • javascriptTextTransformReloadIntervalMinutes: UDF を再読み込みする頻度を指定します(分単位)。値が 0 より大きい場合、Dataflow は Cloud Storage 内の UDF ファイルを定期的にチェックし、ファイルが変更された場合は UDF を再読み込みします。このパラメータを使用すると、パイプラインの実行中にジョブを再起動することなく、UDF を更新できます。値が 0 の場合、UDF の再読み込みは無効になります。デフォルト値は 0 です。

ユーザー定義関数

必要であれば、ユーザー定義関数(UDF)を記述して、このテンプレートを拡張できます。このテンプレートは入力要素ごとに UDF を呼び出します。要素のペイロードは、JSON 文字列としてシリアル化されます。詳細については、Dataflow テンプレートのユーザー定義関数を作成するをご覧ください。

関数の仕様

UDF の仕様は次のとおりです。

  • 入力: JSON 文字列
  • 出力: 文字列または文字列化された JSON オブジェクト。

テンプレートを実行する

コンソール

  1. Dataflow の [テンプレートからジョブを作成] ページに移動します。
  2. [テンプレートからジョブを作成] に移動
  3. [ジョブ名] フィールドに、固有のジョブ名を入力します。
  4. (省略可)[リージョン エンドポイント] で、プルダウン メニューから値を選択します。デフォルトのリージョンは us-central1 です。

    Dataflow ジョブを実行できるリージョンのリストについては、Dataflow のロケーションをご覧ください。

  5. [Dataflow テンプレート] プルダウン メニューから、[ the Pub/Sub to Redis template] を選択します。
  6. 表示されたパラメータ フィールドに、パラメータ値を入力します。
  7. [ジョブを実行] をクリックします。

gcloud

シェルまたはターミナルで、テンプレートを実行します。

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Cloud_PubSub_to_Redis \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       inputSubscription=INPUT_SUBSCRIPTION,\
       redisHost=REDIS_HOST,\
       redisPort=REDIS_PORT,\
       redisPassword=REDIS_PASSWORD,\

次のように置き換えます。

  • JOB_NAME: 一意の任意のジョブ名
  • VERSION: 使用するテンプレートのバージョン

    使用できる値は次のとおりです。

    • latest: 最新バージョンのテンプレートを使用します。このテンプレートは、バケット内で日付のない親フォルダ(gs://dataflow-templates-REGION_NAME/latest/)にあります。
    • バージョン名(例: 2023-09-12-00_RC00)。特定のバージョンのテンプレートを使用します。このテンプレートは、バケット内で対応する日付の親フォルダ(gs://dataflow-templates-REGION_NAME/)にあります。
  • REGION_NAME: Dataflow ジョブをデプロイするリージョン(例: us-central1
  • INPUT_SUBSCRIPTION: Pub/Sub 入力サブスクリプション
  • REDIS_HOST: Redis DB ホスト
  • REDIS_PORT: Redis DB ポート
  • REDIS_PASSWORD: Redis DB のパスワード

API

REST API を使用してテンプレートを実行するには、HTTP POST リクエストを送信します。API とその認証スコープの詳細については、projects.templates.launch をご覧ください。

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "inputSubscription": "INPUT_SUBSCRIPTION",
       "redisHost": "REDIS_HOST",
       "redisPort": "REDIS_PORT",
       "redisPassword": "REDIS_PASSWORD",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Cloud_PubSub_to_Redis",
     "environment": { "maxWorkers": "10" }
  }
}

次のように置き換えます。

  • PROJECT_ID: Dataflow ジョブを実行する Google Cloud プロジェクトの ID
  • JOB_NAME: 一意の任意のジョブ名
  • VERSION: 使用するテンプレートのバージョン

    使用できる値は次のとおりです。

    • latest: 最新バージョンのテンプレートを使用します。このテンプレートは、バケット内で日付のない親フォルダ(gs://dataflow-templates-REGION_NAME/latest/)にあります。
    • バージョン名(例: 2023-09-12-00_RC00)。特定のバージョンのテンプレートを使用します。このテンプレートは、バケット内で対応する日付の親フォルダ(gs://dataflow-templates-REGION_NAME/)にあります。
  • LOCATION: Dataflow ジョブをデプロイするリージョン(例: us-central1
  • INPUT_SUBSCRIPTION: Pub/Sub 入力サブスクリプション
  • REDIS_HOST: Redis DB ホスト
  • REDIS_PORT: Redis DB ポート
  • REDIS_PASSWORD: Redis DB のパスワード

次のステップ