VVR 11 の Hologres コネクタパラメータ explained - Realtime Compute for Apache Flink - Alibaba Cloud - Realtime Compute for Apache Flink

このトピックでは、Ververica Runtime (VVR) バージョン 11 以降で使用可能な WITH パラメーターを一覧表示します。

パラメーターの非推奨化

システムアーキテクチャーおよびメンテナンスの改善を目的として、VVR 8 以前のバージョンで使用されていた一部のレガシーパラメーターが非推奨となりました。以下の表に、これらのパラメーターとその代替パラメーターを示します。

削除されたパラメーター

パラメーター	説明	備考
jdbcRetrySleepInitMs	各リトライ時の固定待機時間です。	増分待機時間を使用してください：`retry-sleep-step-ms`。
jdbcMetaAutoRefreshFactor	キャッシュの残存寿命が指定されたしきい値を下回ると、自動的にキャッシュをリフレッシュします。	`meta-cache-ttl-ms` パラメーターのみを構成すれば十分です。このパラメーターを構成する必要はありません。
type-mapping.timestamp-converting.legacy	Flink と Hologres 間での時刻型の変換を行うかどうかを指定します。	TIMESTAMP_LTZ 型との下位互換性のために導入されたパラメーターであり、現在は不要です。
property-version	コネクターパラメーターのバージョンです。	共通パラメーターのデフォルト値が最適化されたため、削除されました。
field_delimiter	データエクスポート時に行間で使用されるデリミタです。	データ読み取り方法が最適化されたため、削除されました。
jdbcBinlogSlotName	JDBC モードでのバイナリログソーステーブルのスロット名です。	データ読み取り方法が最適化されたため、削除されました。
binlogMaxRetryTimes	エラー発生後のバイナリログデータ読み取りリトライ回数です。	`retry-count` パラメーターを使用して構成できます。
cdcMode	バイナリログデータの読み取りに CDC モードを使用するかどうかを指定します。	デフォルトでは CDC モードでデータが読み取られるため、このパラメーターは削除されました。CDC モード以外を使用する場合は、`source.binlog.change-log-mode` パラメーターを使用して構成してください。
upsertSource	ソーステーブルがアップサート Changelog を使用するかどうかを指定します。	`source.binlog.change-log-mode` パラメーターを構成してください。
bulkload	書き込みに bulkload を使用するかどうかを指定します。	`sink.write-mode` パラメーターを使用してください。
useRpcMode	RPC 経由で Hologres コネクターを使用するかどうかを指定します。	JDBC 接続を使用し、`sink.deduplication.enabled` パラメーターを構成して重複排除を有効または無効にしてください。
partitionrouter	パーティションテーブルへの書き込みを行うかどうかを指定します。	パーティションテーブルへの書き込みがデフォルトでサポートされるようになったため、削除されました。
ignoredelete	リトラクションメッセージを無視するかどうかを指定します。	`sink.delete-strategy` パラメーターを構成してください。このパラメーターは、リトラクションメッセージの処理戦略を指定します。
sdkMode	データの読み取りまたは書き込みに使用する SDK モードを指定します。	このパラメーターは最適化されました。テーブルタイプおよび `source.binlog.read-mode` と `sink.write-mode` パラメーターに基づいて構成してください。
jdbcReadBatchQueueSize	ディメンションテーブル参照リクエスト用のバッファーキューのサイズです。	ポイントクエリのパフォーマンスが低い場合は、`connection.pool.size` パラメーターの構成を推奨します。
jdbcReadRetryCount	ディメンションテーブル参照がタイムアウトした場合のリトライ回数です。	汎用リトライメカニズムの `retry-count` パラメーターは統合されました。
jdbcScanTransactionSessionTimeoutSeconds	スキャン操作のトランザクションタイムアウトです。	一般的なスキャンタイムアウト設定は、`source.scan.timeout-seconds` パラメーターに統合されました。

名称変更されたパラメーター

旧パラメーター	新パラメーター	説明
jdbcRetryCount	retry-count	接続障害発生後の書き込みおよびクエリ操作のリトライ回数です。
jdbcRetrySleepStepMs	retry-sleep-step-ms	各リトライ時の増分待機時間です。
jdbcConnectionMaxIdleMs	connection.max-idle-ms	JDBC 接続の最大アイドル時間です。
jdbcMetaCacheTTL	meta-cache-ttl-ms	ローカルにキャッシュされたテーブルスキーマ情報の生存時間 (TTL) です。
binlog	source.binlog	バイナリログデータの消費を行うかどうかを指定します。
sdkMode	source.binlog.read-mode	読み取りモードを指定します。
binlogRetryIntervalMs	source.binlog.request-timeout-ms	バイナリログ読み取り失敗時のリトライ間隔です。
binlogBatchReadSize	source.binlog.batch-size	バイナリログから一度に読み取る行数です。
binlogStartupMode	source.binlog.startup-mode	バイナリログデータの消費モードです。
jdbcScanFetchSize	source.scan.fetch-size	スキャン時のデータフェッチのバッチサイズです。
jdbcScanTimeoutSeconds	source.scan.timeout-seconds	スキャン操作のタイムアウトです。
enable_filter_push_down	source.scan.filter-push-down.enabled	完全読み取りフェーズ中にフィルタープッシュダウンを有効にするかどうかを指定します。
partition-binlog.mode	source.binlog.partition-binlog-mode	パーティションテーブルからのバイナリログの消費モードです。
partition-binlog-lateness-timeout-minutes	source.binlog.partition-binlog-lateness-timeout-minutes	`DYNAMIC` モードでパーティションテーブルから消費する際の最大遅延タイムアウトです。
partition-values-to-read	source.binlog.partition-values-to-read	`STATIC` モードでパーティションテーブルから消費する際のパーティションを指定します。複数のパーティション値はカンマ (,) で区切ります。
sdkMode	sink.write-mode	書き込みモードを指定します。
mutatetype	sink.on-conflict-action	プライマリキー競合時の処理戦略です。
createparttable	sink.create-missing-partition	パーティションテーブルへの書き込み時に、パーティション値に基づいて存在しないパーティションを自動作成するかどうかを指定します。
jdbcWriteBatchSize	sink.insert.batch-size	Hologres 結果テーブルでのバッチ処理の最大レコード数です。
jdbcWriteBatchByteSize	sink.insert.batch-byte-size	Hologres 結果テーブルでのバッチ処理の最大バイトサイズです。
jdbcWriteFlushInterval	sink.insert.flush-interval-ms	Hologres にデータのバッチをフラッシュするまでの最大待機時間です。
ignoreNullWhenUpdate	sink.ignore-null-when-update.enabled	`sink.on-conflict-action` が更新アクションに設定されている場合、書き込まれるデータ内の null 値を無視するかどうかを指定します。
jdbcEnableDefaultForNotNullColumn	sink.default-for-not-null-column.enabled	`null` がデフォルト値が定義されていない `NOT NULL` 列に書き込まれる場合に、コネクターがデフォルト値を提供するかどうかを指定します。
remove-u0000-in-text.enabled	sink.remove-u0000-in-text.enabled	書き込み時に文字列データから不正な `\u0000` 文字をコネクターが自動的に削除するかどうかを指定します。
partial-insert.enabled	sink.partial-insert.enabled	INSERT 文で定義されたフィールドのみを挿入するかどうかを指定します。
deduplication.enabled	sink.deduplication.enabled	書き込みバッチ処理中に重複排除を実行するかどうかを指定します。
check-and-put.column	sink.insert.check-and-put.column	条件付き更新を有効にし、チェック対象の列を指定します。
check-and-put.operator	sink.insert.check-and-put.operator	条件付き更新操作の比較演算子です。
check-and-put.null-as	sink.insert.check-and-put.null-as	条件付き更新時に既存データが null の場合、null 値をこのパラメーターで指定された値として扱います。
aggressive.enabled	sink.aggressive-flush.enabled	アグレッシブフラッシュモードを有効にするかどうかを指定します。
connectionSize	connection.pool.size	単一の Flink ディメンションテーブルタスク用の JDBC 接続プールのサイズです。
connectionPoolName	connection.pool.name	接続プールの名前です。同一の TaskManager 内で同じプール名を指定したテーブルは、接続プールを共有できます。
jdbcReadBatchSize	lookup.read.batch-size	ディメンションテーブル参照のバッチ処理における最大レコード数です。
jdbcReadTimeoutMs	lookup.read.timeout-ms	ディメンションテーブル参照のタイムアウトです。

WITH パラメーター

一般

パラメーター	説明	型	必須	デフォルト	備考
connector	テーブルのタイプです。	String	はい	—	値は `hologres` である必要があります。
dbname	データベース名です。	String	はい	—	仮想ウェアハウスに接続するには、データベース名にサフィックスを追加します。たとえば、ディメンションテーブルを `read_warehouse` 仮想ウェアハウスに接続するには、`'dbname' = 'db_test@read_warehouse'`
tablename	テーブル名です。	String	はい	—	スキーマが `public` でない場合は、`schema.tableName` 形式を使用します。
username	カスタムアカウントのユーザー名（形式：`BASIC$<user_name>`）。 Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。	String	はい	—	指定されたユーザーは、対応する Hologres データベースへのアクセス権限を持っている必要があります。Hologres の権限およびユーザー管理の詳細については、「開発権限モデル」および「ユーザーの管理」をご参照ください。 AccessKey ID および AccessKey Secret の取得方法については、「AccessKey ID および AccessKey Secret の確認方法」をご参照ください。重要 AccessKey 情報が公開されないようにするため、名前空間変数を使用して AccessKey の値を保存することを推奨します。詳細については、「名前空間変数」をご参照ください。
password	カスタムアカウントのパスワード。 Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret。	String	はい	—
endpoint	Hologres サービスのエンドポイントです。	String	はい	—	詳細については、「エンドポイント」をご参照ください。
connection.pool.size	タスク内の単一の Flink テーブル用に作成される JDBC 接続プールのサイズです。	Integer	いいえ	5	ジョブのパフォーマンスが不十分な場合は、接続プールサイズを増やすことを検討してください。プールサイズはデータスループットに比例します。このパラメーターはディメンションテーブルおよび結果テーブルにのみ影響します。
connection.pool.name	同一の TaskManager 内のテーブルは、同じ名前を使用することで接続プールを共有できます。	String	いいえ	`'default'`	デフォルト値は `'default'` です。複数のテーブルが同じ接続プールを使用するように構成されている場合、それらの中で指定された `connection.pool.size` パラメーターの最大値が使用されます。たとえば、ジョブに 5 つの Hologres テーブル（ディメンションテーブル A および B、結果テーブル C、D、E）が含まれている場合、テーブル A および B には `pool1` を、テーブル C および D には `pool2` を、高トラフィックを処理するテーブル E には `pool3` を割り当てることができます。説明テーブルは、エンドポイントやデータベースなどの接続情報が同じ場合にのみ接続プールを共有できます。ジョブに多くのテーブルが含まれている場合、利用可能な接続数が不足し、パフォーマンスが低下する可能性があります。このような場合は、異なるテーブルに異なるプール名を割り当てることを推奨します。
connection.fixed.enabled	軽量接続モードを使用するかどうかを指定します。	Boolean	いいえ	—	Hologres には接続制限があります。Hologres V2.1 以降では、リアルタイム書き込みでこの制限の対象外となる軽量接続がサポートされています。説明このパラメーターのデフォルト値は、ご利用の Hologres インスタンスのバージョンによって異なります。ディメンションテーブルおよび結果テーブルの場合、コネクターは Hologres バージョン 3.0.28 以降で自動的に軽量接続モードを選択します。ディメンションテーブルの場合、軽量接続モードでは JSONB および RoaringBitmap データ型のクエリはサポートされません。
connection.max-idle-ms	JDBC 接続のアイドルタイムアウト（ミリ秒単位）です。	Long	いいえ	60000	接続がこの時間以上アイドル状態になると、接続は閉じられます。必要な場合は、新しい接続が自動的に作成されます。
connection.ssl.mode	転送中の Secure Sockets Layer (SSL) 暗号化を有効にするかどうか、および使用するモードを指定します。	String	いいえ	disable	`disable`（デフォルト）：転送中の暗号化を無効にします。 `require`：SSL を有効にし、データリンクのみを暗号化します。 `verify-ca`：SSL を有効にし、データリンクを暗号化し、CA 証明書を使用して Hologres サーバーの信頼性を検証します。 `verify-full`：SSL を有効にし、データリンクを暗号化し、CA 証明書を使用して Hologres サーバーの信頼性を検証し、証明書内の Common Name (CN) または DNS 名が構成された Hologres エンドポイントと一致するかを検証します。説明 Hologres V2.1 以降では、`verify-ca` および `verify-full` モードがサポートされています。詳細については、「転送中暗号化」をご参照ください。このパラメーターを `verify-ca` または `verify-full` に設定する場合は、`connection.ssl.root-cert.location` パラメーターも設定する必要があります。
connection.ssl.root-cert.location	証明書ベースの暗号化モードに必要な CA 証明書のパスです。	String	いいえ	—	`connection.ssl.mode` が `verify-ca` または `verify-full` に設定されている場合は、CA 証明書のパスを指定する必要があります。Realtime Compute コンソールのアーティファクトの管理機能を使用して証明書ファイルをアップロードできます。アップロード後、ファイルは /flink/usrlib ディレクトリに保存されます。たとえば、CA 証明書ファイル名が `certificate.crt` の場合、このパラメーターを `'/flink/usrlib/certificate.crt'` に設定します。説明 CA 証明書の取得方法については、「CA 証明書のダウンロード」をご参照ください。
retry-count	接続障害発生時の書き込みまたはクエリ操作のリトライ回数です。	Integer	いいえ	10	—
retry-sleep-step-ms	各リトライ時の増分待機時間（ミリ秒単位）です。	Long	いいえ	5000	たとえば、デフォルト値が 5,000（5 秒）の場合、最初のリトライは 5 秒、2 回目のリトライは 10 秒待機します。
meta-cache-ttl-ms	ローカルにキャッシュされた TableSchema 情報の生存時間（TTL）（ミリ秒単位）です。	Long	いいえ	600000	—
serverless-computing.enabled	サーバーレスリソースを使用するかどうかを指定します。	Boolean	いいえ	false	このパラメーターを `true` に設定すると、コネクターはご利用の Hologres インスタンスのリソースではなく、Hologres サーバーレスリソースを使用して読み取りおよび書き込み操作を実行します。このパラメーターはバッチ読み取りおよびバッチインポートでのみサポートされ、バイナリログ消費、ディメンションテーブルポイントクエリ、リアルタイム書き込みでは有効ではありません。詳細については、「概要」をご参照ください。説明バッチ読み取りは、`source.binlog` が `false` に設定されている場合、または `source.binlog.startup-mode` が `INITIAL` に設定されている場合の完全データ読み取りフェーズで適用されます。バッチインポート操作は、`sink.write-mode` が `COPY_BULK_LOAD` または `COPY_BULK_LOAD_ON_CONFLICT` に設定されている場合に適用されます。説明大規模なデータのインポートまたはエクスポートを行う場合は、ご利用の Hologres インスタンス上の他のクエリへの影響を避けるために、このパラメーターを有効にすることを推奨します。詳細については、「概要」をご参照ください。

ソーステーブルパラメーター

パラメーター	説明	型	必須	デフォルト	備考
source.binlog	バイナリログデータを消費するかどうかを指定します。	Boolean	いいえ	true	`true`（デフォルト）：バイナリログデータを消費します。 `false`：バイナリログデータを消費しません。ジョブはバッチ読み取りを実行して停止します。
source.binlog.read-mode	読み取りモードを指定します。	ENUM	いいえ	AUTO	`AUTO`（デフォルト）：インスタンスバージョンに基づいて最適なモードを自動選択します。 `HOLOHUB`：HoloHub モードを使用してバイナリログデータを消費します。 `JDBC`：JDBC モードを使用してバイナリログデータを消費します。説明 AUTO モードの選択ロジックは次のとおりです。 Hologres インスタンス V2.1.27 以降の場合、コネクターは JDBC モードを選択し、軽量接続がデフォルトで有効になります（`connection.fixed.enabled` パラメーターのデフォルト値は `true` です）。 Hologres インスタンス V2.1.0 ～ V2.1.26 の場合、コネクターは JDBC モードを選択します。 Hologres インスタンス V2.0 以前の場合、コネクターは HOLOHUB モードを選択します。
source.binlog.change-log-mode	CDC ソーステーブルでサポートされる ChangeLog のタイプを指定します。	ENUM	いいえ	UPSERT	`ALL`：INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER を含むすべての ChangeLog タイプをサポートします。 `UPSERT`（デフォルト）：INSERT、DELETE、UPDATE_AFTER を含むアップサート ChangeLog のみをサポートします。 `ALL_AS_APPEND_ONLY`：すべての ChangeLog タイプを INSERT として扱います。説明ダウンストリームパイプラインにリトラクションオペレーターが含まれている場合（たとえば、`ROW_NUMBER OVER WINDOW` を使用して重複排除を行う場合）、`upsertSource` を `true` に設定する必要があります。この場合、ソーステーブルは Hologres からアップサートモードでデータを読み取ります。
source.binlog.startup-mode	バイナリログデータの消費モードを指定します。	ENUM	いいえ	INITIAL	`INITIAL`（デフォルト）：データの完全読み取りを実行した後、バイナリログからの増分消費を開始します。 `EARLIEST_OFFSET`：利用可能な最も古いバイナリログオフセットから消費を開始します。 `TIMESTAMP`：指定された `startTime` のバイナリログから消費を開始します。 `LATEST_OFFSET`：最新のバイナリログオフセットから消費を開始します。説明 startTime が設定されている場合、または起動インターフェイスで開始時刻が選択されている場合、このパラメーターは自動的に `TIMESTAMP` モードに設定され、他の消費モードは無視されます。startTime パラメーターが優先されます。 `LATEST_OFFSET` モードは、VVR 11.6 以降でのみサポートされています。
source.binlog.batch-size	バイナリログから 1 回のバッチで読み取る行数を指定します。	Integer	いいえ	512	該当なし。
source.binlog.request-timeout-ms	バイナリログデータの読み取りタイムアウトを指定します。	Long	いいえ	300000	単位：ミリ秒。説明タイムアウトは、ダウンストリームオペレーターがソーステーブルデータを処理する速度が遅すぎてバックプレッシャーが発生している可能性を示しています。
source.binlog.project-columns.enabled	バイナリログデータの読み取り時に、ユーザーテーブルで指定されたフィールドのみを読み取るかどうかを指定します。	Boolean	いいえ	なし	指定されたフィールドとは、`CREATE TEMPORARY TABLE` 文で宣言されたフィールドです。宣言されていないフィールドは読み取られません。テーブルに多くのフィールドがあるが、その一部のみを消費する必要がある場合、このオプションにより不要なデータ転送および変換を回避でき、読み取りパフォーマンスが向上し、帯域幅が節約されます。説明このパラメーターは、VVR 11.3 以降および Hologres インスタンス V3.2 以降でのみサポートされています。通常、このパラメーターを構成する必要はありません。バージョン要件を満たす場合、コネクターがデフォルトで有効にします。
source.binlog.compression.enabled	バイナリログデータの読み取り時に転送中のデータ圧縮を有効にするかどうかを指定します。	Boolean	いいえ	なし	バイナリログを消費する際、サーバーは LZ4 アルゴリズムで圧縮されたバイトストリームを返します。これにより、読み取りパフォーマンスが向上し、帯域幅使用量が削減されます。説明このパラメーターは、VVR 11.3 以降および Hologres インスタンス V3.2 以降でのみサポートされています。通常、このパラメーターを構成する必要はありません。バージョン要件を満たす場合、コネクターがデフォルトで有効にします。
source.binlog.partition-binlog-mode	パーティションテーブルのバイナリログ消費モードを指定します。	Enum	いいえ	DISABLE	`DISABLE`（デフォルト）：非パーティションソーステーブルに使用します。指定された Hologres テーブルがパーティションテーブルの場合、ジョブは例外をスローします。 `DYNAMIC`：パーティションテーブルの最新パーティションを継続的に消費します。テーブルには動的パーティショニングが有効になっている必要があります。DYNAMIC モードでは、パーティションを時系列順に消費します。最新から 2 番目のパーティションの消費に到達すると、新しい時間単位が始まった時点で最新パーティションの消費を開始します。 `STATIC`：パーティションテーブルから固定されたパーティションセットを消費します。複数のパーティションを同時に消費できます。消費中にパーティションを追加または削除することはできません。デフォルトでは、親テーブルのすべてのパーティションを消費します。
source.binlog.partition-binlog-lateness-timeout-minutes	DYNAMIC モードで、パーティションテーブルを消費する際の最大遅延タイムアウトを指定します。	Integer	いいえ	60	単位：分。DYNAMIC モードでは、新しい時間単位が始まった時点で、コネクターは現在時刻に対応する最新パーティションの消費を開始します。ただし、直前のパーティションをすぐに閉じるのではなく、遅れて到着するデータをキャプチャするために監視を続けます。たとえば、動的パーティショニングが DAY 単位で、パーティションが `20240920` の場合、データの最大遅延が 1 時間であれば、このパーティションの消費は 2024-09-21 00:00:00 ではなく、2024-09-21 01:00:00 に停止します。 `lateness-timeout` 値は、パーティションの時間単位を超えることはできません。パーティショニングが日単位の場合、最大値は 24 * 60 = 1440 分です。DYNAMIC モードでは、通常は一度に 1 つのパーティションしか消費しませんが、遅延期間中は 2 つのパーティションを同時に消費する場合があります。
source.binlog.partition-values-to-read	STATIC モードで、消費するパーティションを指定します。パーティション値はカンマ (,) で区切ります。	String	いいえ	なし	このパラメーターを構成しない場合、STATIC モードでは指定された親テーブルのすべてのパーティションを消費します。構成した場合は、指定されたパーティションのみを消費します。パーティション名全体ではなく、パーティション値のみを指定します。複数の値はカンマ (`,`) で区切ります。このパラメーターは正規表現をサポートしていません。
startTime	消費オフセットの開始時刻を指定します。	String	いいえ	なし	形式：yyyy-MM-dd hh:mm:ss。このパラメーターが設定されておらず、ジョブが保存された状態から再開されない場合、消費は最も古いバイナリログから開始されます。
source.scan.fetch-size	バッチ読み取りのバッチサイズを指定します。	Integer	いいえ	512	該当なし。
source.scan.timeout-seconds	バッチ読み取りのタイムアウトを指定します。	Integer	いいえ	60	単位：秒。
source.scan.filter-push-down.enabled	バッチ読み取り中にフィルターをプッシュダウンするかどうかを指定します。	Boolean	いいえ	false	`false`（デフォルト）：フィルターをプッシュダウンしません。 `true`：バッチ読み取り中に、サポートされているフィルター条件を Hologres にプッシュダウンします。説明このパラメーターは、`source.binlog.filter-push-down.enabled` パラメーターと同時に有効にすることはできません。このパラメーターは次の 2 つのシナリオで有効になります。 `source.binlog` が `false` に設定されている場合、バッチ読み取りが実行され、フィルタープッシュダウンが有効になります。 `source.binlog` を true に設定し、`source.binlog.startup-mode` を `INITIAL` に設定すると、完全データおよび増分データの読み取りが有効になり、フィルタープッシュダウンは完全データ読み取りフェーズで有効になります。
source.binlog.filter-push-down.enabled	バイナリログ消費中にフィルターをプッシュダウンするかどうかを指定します。	Boolean	いいえ	false	`false`（デフォルト）：フィルターをプッシュダウンしません。 `true`：バイナリログ消費中に、サポートされているフィルター条件を Hologres にプッシュダウンします。説明このパラメーターは、VVR 11.3 以降および Hologres インスタンスバージョン 4.0 以降でのみサポートされています。このパラメーターは、`source.scan.filter-push-down.enabled` パラメーターと同時に有効にすることはできません。 `source.binlog` が true に設定されている場合、フィルタープッシュダウンが有効になります。たとえば、`source.binlog.startup-mode` が `INITIAL` に設定されている場合、フィルタープッシュダウンは完全データおよび増分フェーズの両方で有効になります。
scan.prefer.physical-column.over.metadata-column	物理列とメタデータ列の名前が同じ場合に、物理列からのデータ読み取りを優先するかどうかを指定します。	Boolean	いいえ	false	このパラメーターは、VVR 11.5 以降でのみサポートされています。それ以前のバージョンでは、常にメタデータ列からのデータ読み取りが優先されます。

シンクテーブルのパラメーター

パラメーター	説明	型	必須	デフォルト	備考
sink.write-mode	データ書き込みモードです。	ENUM	いいえ	INSERT	`INSERT`：デフォルト。INSERT 文を使用して JDBC でデータを書き込みます。 `COPY_STREAM`：固定 COPY ストリーミング方式を使用します。これは、高スループットおよび低遅延が求められるシナリオ向けのパフォーマンス専有型オプションです。このモードでは、データの削除、親パーティションテーブルへの書き込み、`sink.ignore-null-when-update.enabled` パラメーターの使用はサポートされていません。 `COPY_BULK_LOAD`：COPY プロトコルを使用してバルク書き込みを実行します。`COPY_BULK_LOAD` は現在、プライマリキーを持たないテーブルにのみ適用されます（プライマリキーが重複している場合は例外がスローされます）。`COPY_STREAM` と比較して、このモードでは Hologres の書き込みリソースをより少なく使用します。 `COPY_BULK_LOAD_ON_CONFLICT`：COPY プロトコルを使用してバッチ書き込みを実行し、プライマリキーを持つテーブルへの書き込みおよびプライマリキー競合の処理をサポートします。説明 `COPY_BULK_LOAD_ON_CONFLICT` モードは、VVR 11.3 以降および Hologres V3.1 以降でのみサポートされています。このモードでは、Flink 側で Hologres 結果テーブルの DistributionKey に基づいてデータをリシャッフルします。これにより、同じシャードのデータが同じ Flink タスクによって書き込まれるようになり、バッチインポート時のテーブルロックの範囲がテーブルレベルからシャードレベルに縮小され、異なるシャードへの同時書き込みが可能になります。したがって、ジョブの同時実行数を Hologres 結果テーブルのShard Count に合わせて設定してください。 `COPY_BULK_LOAD` および `COPY_BULK_LOAD_ON_CONFLICT` モードでは、チェックポイントが成功した後にのみデータが可視になります。これらのモードは、即時のデータ可視性が不要なシナリオや、既存データのバッチインポートに適しています。 `COPY_STREAM` モードでは、`CURRENT_TIMESTAMP` や `NOW()` などの時間関数の値は、COPY 接続が確立された時点で固定され、各レコードごとに更新されません。Hologres テーブル列でこれらの関数をデフォルト値として使用している場合（例：`DEFAULT CURRENT_TIMESTAMP`）、列の値はレコードの実際の書き込み時刻ではなく、接続確立時刻を反映します。正確な書き込み時刻を取得するには、Hologres のデフォルト値に依存せず、Flink ジョブ内でタイムスタンプ値を明示的に指定する必要があります。
sink.on-conflict-action	プライマリキー競合時のポリシーです。	ENUM	いいえ	INSERT_OR_UPDATE	`INSERT_OR_IGNORE`：最初のレコードを保持し、同じプライマリキーを持つ後続のすべてのレコードを無視します。 `INSERT_OR_REPLACE`：既存の行を新しい行で完全に置き換えます。 `INSERT_OR_UPDATE`：デフォルト。既存の行の一部の列のみを更新します。たとえば、列 a、b、c、d を持つテーブルがあり、a がプライマリキーであるとします。着信レコードが列 a および b のみの値を提供する場合、プライマリキー競合により列 b のみが更新され、列 c および d は変更されません。
sink.create-missing-partition	パーティションテーブルへの書き込み時に、一致するパーティションが存在しない場合にパーティション値に基づいてパーティションを自動作成するかどうかを指定します。	Boolean	いいえ	false	パーティションキーとして DATE データ型を使用し、動的パーティショニングを有効にしている場合、自動作成されたパーティションの名前は動的パーティショニングの命名規則に従います。このパラメーターは慎重に使用してください。パーティション値にダーティデータが含まれていないことを確認してください。そうでないと、誤ったパーティションが作成され、フェールオーバーが発生する可能性があります。 `sink.write-mode` が `INSERT` でない場合、親パーティションテーブルへの書き込みはできません。
sink.delete-strategy	リトラクションメッセージの処理ポリシーです。	String	いいえ	CHANGELOG_STANDARD	`IGNORE_DELETE`：UPDATE_BEFORE および DELETE メッセージを無視します。このオプションは、削除ではなく挿入または更新のみが必要なシナリオに適しています。 `NON_PK_FIELD_TO_NULL`：UPDATE_BEFORE メッセージを無視し、DELETE メッセージを処理してプライマリキー以外のフィールドを NULL に設定します。このオプションは、他の列に影響を与えずに削除を実行したい部分更新シナリオに適しています。 `DELETE_ROW_ON_PK`：UPDATE_BEFORE メッセージを無視し、DELETE メッセージを処理してプライマリキーに基づいて行全体を削除します。このオプションは、行全体を削除したい部分更新シナリオに適しています。 `CHANGELOG_STANDARD`：標準の Flink SQL changelog メカニズムに従います。削除操作を無視しません。このオプションでは、古いレコードを削除してから新しいレコードを挿入することで更新を実行し、データ精度を確保します。このオプションは、部分更新を含まないシナリオに適しています。説明 `NON_PK_FIELD_TO_NULL` オプションを有効にすると、プライマリキー列のみに値が設定され、他のすべての列が null になるレコードが生成される可能性があります。
sink.ignore-null-when-update.enabled	`sink.on-conflict-action` が 'INSERT_OR_UPDATE' の場合、更新時の着信データ内の null 値を無視するかどうかを指定します。	Boolean	いいえ	false	`false`：デフォルト。null 値を Hologres 結果テーブルに書き込みます。 `true`：着信更新データ内の null 値を無視します。説明このパラメーターは、`sink.write-mode` が `INSERT` に設定されている場合にのみサポートされます。
sink.ignore-null-when-update-by-expr.enabled	`sink.on-conflict-action` が 'INSERT_OR_UPDATE' の場合、式ベースの方法を使用して着信更新データ内の null 値を無視するかどうかを指定します。	Boolean	いいえ	false	この方法は、`sink.ignore-null-when-update.enabled` よりもパフォーマンスが優れています。 `false`：デフォルト。 `sink.ignore-null-when-update.enabled` が `true` の場合、更新時の null 値は無視されます。 `sink.ignore-null-when-update.enabled` が `false` の場合、null 値は Hologres 結果テーブルに書き込まれます。 `true`：着信更新データ内の null 値を無視します。`sink.ignore-null-when-update.enabled` の設定に関係なく適用されます。説明このパラメーターは、`sink.write-mode` が `INSERT` に設定されている場合にのみサポートされます。 Hologres V4.0 以降が必要です。
sink.default-for-not-null-column.enabled	null がデフォルト値が定義されていない `NOT NULL` 列に書き込まれる場合に、コネクターがデフォルト値を提供するかどうかを指定します。	Boolean	いいえ	true	`true`：デフォルト。次のルールに基づいてデフォルト値を提供および書き込みます。 String データ型の場合、空文字列 (`""`) を書き込みます。 Number データ型の場合、`0` を書き込みます。 Date、timestamp、または timestamptz データ型の場合、1970-01-01 00:00:00 を書き込みます。 `false`：デフォルト値を提供しません。`NOT NULL` 列に null 値を書き込むと、例外がスローされます。説明このパラメーターは、`sink.write-mode` が `INSERT` に設定されており、かつ `sink.on-conflict-action` が `INSERT_OR_UPDATE` 以外のオプションに設定されている場合にのみサポートされます。
sink.remove-u0000-in-text.enabled	書き込み操作中に文字列型から不正な `\u0000` 文字をコネクターが削除するかどうかを指定します。	Boolean	いいえ	true	`false`：コネクターはデータを変更しません。ダーティデータが存在する場合、書き込み操作で次のようなエラーがスローされる可能性があります：`ERROR: invalid byte sequence for encoding "UTF8": 0x00` この場合、ソーステーブルでダーティデータを処理するか、SQL ステートメントでダーティデータの処理ロジックを定義する必要があります。 `true`：デフォルト。`\u0000` 文字を文字列型から削除して、書き込みエラーを防止します。
sink.partial-insert.enabled	INSERT 文で定義されたフィールドのみを挿入または更新するかどうかを指定します。	Boolean	いいえ	false	`false`：デフォルト。結果テーブルの DDL で定義されたすべてのフィールドが更新されます。INSERT 文で宣言されていないフィールドは null に更新されます。 `true`：INSERT 文で定義されたフィールドのみをコネクターにプッシュし、宣言されたフィールドのみを更新または挿入できます。説明このパラメーターは、`sink.on-conflict-action` パラメーターが `INSERT_OR_UPDATE` に設定されている場合にのみ有効になります。
sink.deduplication.enabled	バッチ処理中に重複排除を実行するかどうかを指定します。	Boolean	いいえ	true	`true`：デフォルト。バッチに同じプライマリキーを持つ複数のレコードが含まれている場合、コネクターはそれらを重複排除し、最後のレコードのみを Hologres 結果テーブルに書き込みます。たとえば、2 つのフィールドを持つテーブルで、最初のフィールドがプライマリキーであるとします。レコード `INSERT (1,'a')` および `INSERT (1,'b')` が順に到着した場合、重複排除後、最後のレコード `(1,'b')` のみが Hologres 結果テーブルに書き込まれます。 Hologres 結果テーブルにすでにレコード `(1,'a')` が含まれており、レコード `DELETE (1,'a')` および `INSERT (1,'b')` が順に到着した場合、最後のレコード `(1,'b')` のみが Hologres に書き込まれます。これにより、削除後に挿入するのではなく、直接更新が行われます。 `false`：バッチ処理中の重複排除を無効にします。新しいレコードが現在のバッチ内のレコードと同じプライマリキーを持つ場合、まずバッチが結果テーブルに書き込まれ、その後新しいレコードが処理されます。説明このパラメーターは、`sink.write-mode` が `INSERT` に設定されている場合にのみサポートされます。重複排除が無効になっている場合、すべてのレコードが同じプライマリキーを持つような極端なケースでは、書き込み操作が単一行書き込みに劣化し、パフォーマンスが低下する可能性があります。
sink.aggressive-flush.enabled	アグレッシブフラッシュモードを有効にするかどうかを指定します。	Boolean	いいえ	false	true に設定すると、バッチが構成されたサイズに達していなくても、接続がアイドル状態になったときにコネクターが強制的にバッチをフラッシュします。これにより、トラフィックが少ない期間のデータ書き込み遅延が短縮されます。説明このパラメーターは、`sink.write-mode` が `INSERT` または `COPY_STREAM` に設定されている場合にのみサポートされます。
sink.insert.check-and-put.column	条件付き更新を有効にし、チェック対象の列を指定します。	String	いいえ	なし	このパラメーターの値は、Hologres テーブル内の既存の列名である必要があります。重要このパラメーターは、`sink.write-mode` が `INSERT` に設定されている場合にのみサポートされます。結果テーブルにはプライマリキーが必要であり、`sink.on-conflict-action` パラメーターは `INSERT_OR_UPDATE` または `INSERT_OR_REPLACE` である必要があります。逆引き参照が必要なため、パフォーマンスを向上させるには、結果テーブルを行指向テーブルまたはハイブリッド行・列テーブルとして作成してください。多くのレコードが同じプライマリキーを共有している場合、check-and-put 操作が単一行書き込みに劣化し、書き込みパフォーマンスが低下する可能性があります。
sink.insert.check-and-put.operator	条件付き更新操作の比較演算子です。	String	いいえ	GREATER	着信レコードのチェック列とテーブル内の既存値を比較します。条件が満たされた場合に更新が実行されます。サポートされている演算子は、`GREATER`、`GREATER_OR_EQUAL`、`EQUAL`、`NOT_EQUAL`、`LESS`、`LESS_OR_EQUAL`、`IS_NULL`、`IS_NOT_NULL` です。
sink.insert.check-and-put.null-as	条件付き更新時に、既存データが null の場合、null 値をこのパラメーターで指定された値として扱います。	String	いいえ	なし	PostgreSQL では、`NULL` との比較はすべて `FALSE` を返します。したがって、テーブル内の既存値が `NULL` の場合、比較に使用する特定の値をこのパラメーターで設定する必要があります。これは SQL の `COALESCE` 関数と同等です。
sink.insert.batch-size	`INSERT` モードで、書き込み前にバッチに含めるレコードの最大数です。	Integer	いいえ	512	コネクターは、`sink.insert.batch-size`、`sink.insert.batch-byte-size`、または `sink.insert.flush-interval-ms` のいずれかのしきい値に達した時点で書き込みをトリガーします。
sink.insert.batch-byte-size	`INSERT` モードで、書き込み前のバッチの最大サイズ（バイト単位）です。	Long	いいえ	2097152 (2 MB)
sink.insert.flush-interval-ms	`INSERT` モードで、Hologres にバッチをフラッシュするまでの最大間隔（ミリ秒単位）です。	Long	いいえ	10000
sink.copy.format	COPY モードで使用されるデータ転送フォーマットです。	String	いいえ	`COPY_STREAM` モードのデフォルトは `binary` です。 `COPY_BULK_LOAD` または `COPY_BULK_LOAD_ON_CONFLICT` モードのデフォルトは `text` です。	`COPY_STREAM` モードでサポートされているフォーマット： binary text binaryrow（Hologres エンジン V4.1.0 以降が必要） `COPY_BULK_LOAD` または `COPY_BULK_LOAD_ON_CONFLICT` モードでサポートされているフォーマット：`text` のみ説明このパラメーターは、`sink.write-mode` が `COPY_STREAM`、`COPY_BULK_LOAD`、または `COPY_BULK_LOAD_ON_CONFLICT` に設定されている場合にのみサポートされます。
sink.insert.conflict-update-set	プライマリキー競合時に行を更新するために使用される Hologres 式です。	String	いいえ	なし	`insert into tbl values(xxx) on conflict(pk) do update set <conflict-update-set>` と同等です。Hologres 式または関数を指定できます。たとえば、このパラメーターを `col1=old.col1+excluded.col1,col2=excluded.col2` に設定すると、プライマリキー競合時に `col1` の値が古い値と新しい値の合計に更新され、`col2` の値が新しい値に更新されます。このパラメーターを指定しない場合、コネクターはデフォルトで、着信フィールドのすべてを新しい値に更新します。古い値に依存するステートフルな式（例：`col=old.col+excluded.col`）の場合、列をレコードバージョン番号として使用できるようにし、`sink.insert.conflict-where` を `excluded.seq>old.seq` に設定してください。これにより、フェールオーバーおよび回復後のデータ精度が保証されます。説明このパラメーターは、`sink.write-mode` が `INSERT` に設定されている場合にのみサポートされます。
sink.insert.conflict-where	プライマリキー競合時に更新をトリガーするための Hologres フィルター条件です。	String	いいえ	なし	`insert into tbl values(xxx) on conflict(pk) do update set <conflict-update-set> where <conflict-where>` と同等です。Hologres 式または関数を指定できます。たとえば、このパラメーターを `excluded.col1>old.col1` に設定すると、`col1` の新しい値が既存値より大きい場合にのみ更新がトリガーされます。説明このパラメーターは、`sink.write-mode` が `INSERT` に設定されている場合にのみサポートされます。このパラメーターは `sink.insert.check-and-put*` パラメーターと競合します。これらを同時に構成するとエラーが発生します。

ディメンションテーブルパラメーター

パラメーター	説明	型	必須	デフォルト	備考
lookup.read.batch-size	Hologres ディメンションテーブルのポイントクエリでバッチ処理するレコードの最大数です。	Integer	いいえ	256	なし
lookup.read.timeout-ms	ディメンションテーブルのポイントクエリのタイムアウトです。	Long	いいえ	0（タイムアウトなし）	なし
lookup.read.column-table.enabled	ディメンションテーブルとして列指向テーブルを使用するかどうかを指定します。	Boolean	いいえ	false	列指向テーブルはポイントクエリのパフォーマンスが低いため、ディメンションテーブルとしては行指向テーブルまたはハイブリッド行・列テーブルを使用することを推奨します。列指向テーブルに対してこのパラメーターを有効にすると、警告がログに記録されます。
lookup.insert-if-not-exists	レコードが存在しない場合に挿入するかどうかを指定します。	Boolean	いいえ	false	ポイントクエリで現在のレコードに対するディメンションテーブル内の一致が見つからない場合、コネクターはそのレコードを挿入します。
cache	キャッシュポリシーです。	String	いいえ	なし	Hologres では、None および LRU の 2 つのキャッシュポリシーのみがサポートされています。
cacheSize	キャッシュサイズ（行単位）です。	Integer	いいえ	10000	LRU キャッシュポリシーを選択した後、キャッシュサイズを設定できます。単位はエントリー数です。
cacheTTLMs	キャッシュの生存時間（TTL）（ミリ秒単位）です。	Long	いいえ	備考を参照	単位はミリ秒です。cacheTTLMs のデフォルト値はキャッシュ構成によって異なります。キャッシュが LRU に構成されている場合、cacheTTLMs はキャッシュタイムアウトを指定します。デフォルト設定では、期限切れになりません。キャッシュが None に構成されている場合、cacheTTLMs は省略可能で、キャッシュがタイムアウトしないことを示します。
cacheEmpty	空の結合結果をキャッシュするかどうかを指定します。	Boolean	いいえ	true	`true`（デフォルト）：空の結合結果をキャッシュします。 `false`：空の結合結果をキャッシュしません。ただし、結合文の `AND` 前の条件が満たされても `AND` 後の条件が満たされない場合、空の結果が依然としてキャッシュされます。次のコードがその例です。 `LEFT JOIN latest_emergency FOR SYSTEM_TIME AS OF PROCTIME() AS t2 ON t1.alarm_id = t2.alarm_id -- 動的アラームが検出された場合は、動的アラーム ID を使用して一致させます。それ以外の場合は、動的アラーム ID フィールドを無視します。 AND CASE WHEN alarm_type = 2 THEN t1.dynamic_id = t2.dynamic_alarm_id ELSE true END` 重要静的またはゆっくり変化するディメンションテーブルのパフォーマンスを向上させ、I/O 負荷を軽減するには、デフォルト値（`true`）を使用してください。頻繁に更新されるディメンションテーブルの場合は、このパラメーターを `false` に設定するか、短い `cacheTTLMs` 値を使用してください。これにより、空のキャッシュエントリーが迅速に削除され、後続の結合失敗を防ぐことができます。
async	結果を非同期で返すかどうかを指定します。	Boolean	いいえ	false	`true`：結果を非同期で返します。 `false`（デフォルト）：結果を同期的に返します。説明非同期の結果は順不同です。ディメンションテーブルの非同期クエリは、async パラメーターと lookup.async パラメーターの両方によって制御されます。async パラメーターまたは lookup.async パラメーターのいずれかが true の場合、非同期クエリが有効になります。 `lookup.async` のデフォルト値は `true` です。したがって、`async` を `false` に設定しても、非同期クエリは有効になります。非同期クエリを無効にするには、両方のパラメーターを `false` に設定する必要があります。
lookup.async	ディメンションテーブルの非同期クエリを有効にするかどうかを指定します。	Boolean	いいえ	true	`true`（デフォルト）：非同期クエリを有効にします。 `false`：非同期クエリを無効にします。
lookup.filter-push-down.enabled	ディメンションテーブルのフィルター条件を Hologres サーバーにプッシュダウンするかどうかを指定します。	Boolean	いいえ	false	この機能は、列と定数の間の比較演算のみをサポートします。コネクターは、等号 (=) および比較 (<, <=, >, >=) 演算子のみをプッシュダウンします。説明このパラメーターは、Realtime Compute Engine VVR 11.4 以降でのみサポートされています。