RabbitMQ コネクタ リファレンス

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

このページには、LakeFlow Connectのマネージド RabbitMQ コネクタのリファレンスが含まれています。

コネクターオプション

次のオプションで、取り込みパイプライン内の各送信先テーブルの RabbitMQ ソースを構成します。パイプライン定義の connector_options.rabbitmq_options で、これらのオプションを指定してください。完全なパイプラインの例については、「例」を参照してください。

オプション	Type	デフォルト	説明
queue	String	—	(必須) 消費元のRabbitMQクラシックキューの名前

オプション	Type	デフォルト	説明
queue	String	—	(必須) 消費元のRabbitMQクラシックキューの名前

接続プロパティ

カタログエクスプローラで Unity Catalog RabbitMQ 接続を作成するときは、次のプロパティを指定する必要があります。接続作成のステップについては、RabbitMQ 接続を作成するを参照してください。

属性	説明
接続名	Unity Catalog内の接続の一意の名前。
接続タイプ	RabbitMQ を選択してください。
ホスト	RabbitMQブローカーのホスト名。
ユーザー名	RabbitMQ ブローカーとの認証に使用されるユーザー名。
パスワード	RabbitMQブローカーでの認証に使用されるパスワードです。
仮想ホスト	RabbitMQブローカーの仮想ホスト。デフォルトの仮想ホストは/です。
ポート	RabbitMQブローカーのポートです。デフォルトのポートは 5672です。

属性	説明
接続名	Unity Catalog内の接続の一意の名前。
接続タイプ	RabbitMQ を選択してください。
ホスト	RabbitMQブローカーのホスト名。
ユーザー名	RabbitMQ ブローカーとの認証に使用されるユーザー名。
パスワード	RabbitMQブローカーでの認証に使用されるパスワードです。
仮想ホスト	RabbitMQブローカーの仮想ホスト。デフォルトの仮想ホストは/です。
ポート	RabbitMQブローカーのポートです。デフォルトのポートは 5672です。

宛先テーブルスキーマ

RabbitMQコネクタは、ストリーミングテーブル（追記専用）に書き込みます。By default, 各宛先テーブルには単一の列が含まれます：

列	Type	説明
body	BINARY	RabbitMQ メッセージボディの生のバイナリコンテンツ。このコネクタは、取り込み中にボディのデコードやデシリアライズを行いません。必要に応じて下流でデコードします。

列	Type	説明
body	BINARY	RabbitMQ メッセージボディの生のバイナリコンテンツ。このコネクタは、取り込み中にボディのデコードやデシリアライズを行いません。必要に応じて下流でデコードします。

ソースメタデータを含める

RabbitMQ メッセージのメタデータもキャプチャするには、宛先テーブルの table_configuration で source_metadata_column オプションを設定します。設定すると、コネクタはその名前の構造体列を body と並べて追加し、次のフィールドを含めます。

フィールド	Type	説明
exchange	STRING	メッセージが公開された取引所。
routing_key	STRING	メッセージの公開時に使用されたルーティングキー。
queue	STRING	メッセージが消費されたキュー。
redelivered	BOOLEAN	true 確認応答の失敗後にメッセージが再配信された場合。
message_id	STRING	プロデューサーが設定したメッセージID（存在する場合）。
correlation_id	STRING	プロデューサーが設定した相関ID（存在する場合）。
content_type	STRING	プロデューサーが設定したコンテンツタイプ（存在する場合）。
content_encoding	STRING	プロデューサーが設定したコンテンツエンコーディング（存在する場合）。
timestamp	BIGINT	プロデューサーが設定したメッセージのTimestamp (存在する場合)。
headers	MAP<STRING, STRING>	プロデューサーが設定したメッセージヘッダー（存在する場合）。

フィールド	Type	説明
exchange	STRING	メッセージが公開された取引所。
routing_key	STRING	メッセージの公開時に使用されたルーティングキー。
queue	STRING	メッセージが消費されたキュー。
redelivered	BOOLEAN	true 確認応答の失敗後にメッセージが再配信された場合。
message_id	STRING	プロデューサーが設定したメッセージID（存在する場合）。
correlation_id	STRING	プロデューサーが設定した相関ID（存在する場合）。
content_type	STRING	プロデューサーが設定したコンテンツタイプ（存在する場合）。
content_encoding	STRING	プロデューサーが設定したコンテンツエンコーディング（存在する場合）。
timestamp	BIGINT	プロデューサーが設定したメッセージのTimestamp (存在する場合)。
headers	MAP<STRING, STRING>	プロデューサーが設定したメッセージヘッダー（存在する場合）。

選択する名前は body (予約済みの出力列) であってはならず、空であってはならず、スペース、,;{}() のいずれかの文字、改行、tab、または = を含めてはなりません。source_metadata_column が設定されていない場合、コネクタは body 列のみを書き込みます。

パイプラインの例については、「ソースメタデータを持つパイプライン」を参照してください。