Oracle Database Sink Connector for Confluent Cloud¶
注釈
If you are installing the connector locally for Confluent Platform, see JDBC Connector (Source and Sink) for Confluent Platform.
マネージド型の Oracle Database Sink Connector for Confluent Cloud を使用すると、データを Apache Kafka® トピックから、Oracle データベース(JDBC)にエクスポートできます。コネクターによって、Kafka からデータがポーリングされ、トピックのサブスクリプションに基づいてデータベースに書き込まれます。アップサートを使用してべき等性のある書き込みを行うことができます。テーブルの自動作成と、制限付きの自動進化もサポートされています。
機能¶
Oracle Database Sink Connector は、次の機能をサポートしています。
- べき等性のある書き込み: デフォルトの
insert.mode
は INSERT です。UPSERT として構成されている場合、コネクターはプレーンな insert ステートメントではなく upsert セマンティクスを使用します。upsert セマンティクスでは、プライマリキーの制約違反がある場合にアトミックに新しい行を追加したり既存の行をアップデートしたりします。これにより、べき等性が確保されます。 - SSL のサポート: 一方向 SSL をサポートします。
- スキーマ: このコネクターは、Avro、JSON スキーマ、および Protobuf 入力 値 フォーマットをサポートします。このコネクターは、Avro、JSON スキーマ、Protobuf、および String の入力 キー フォーマットをサポートします。スキーマレジストリ ベースのフォーマットを使用するには、Schema Registry を有効にしておく必要があります。
- プライマリキーのサポート: サポートされる PK モード は、
kafka
、none
、record_key
、およびrecord_value
です。PK Fields プロパティと組み合わせて使用します。 - テーブルおよび列の自動作成:
auto.create
およびauto-evolve
がサポートされます。テーブルまたは列がない場合に、自動的に作成することができます。テーブル名は Kafka トピック名に基づいて作成されます。 - 少なくとも 1 回のデリバリー: コネクターによって、Kafka のトピックからのレコードが少なくとも 1 回は配信されることが保証されます。
- 複数のタスクのサポート: このコネクターは、1 つまたは複数のタスクの実行をサポートしています。タスクが多いほどパフォーマンスが向上する可能性があります。
Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。
利用可能なすべての構成プロパティの詳細については、「構成プロパティ」を参照してください。Cloud コネクターの制限事項 も参照してください。
クイックスタート¶
このクイックスタートを使用して、Confluent Cloud Oracle Database Sink Connector の利用を開始することができます。このクイックスタートでは、コネクターを選択し、イベントをストリーミングするようにコネクターを構成するための基本的な方法について説明します。
- 前提条件
- アマゾンウェブサービス (AWS)、Microsoft Azure (Azure)、または Google Cloud Platform (GCP)上の Confluent Cloud クラスターへのアクセスを許可されていること。
- Oracle データベースへのアクセスを許可されていること。
- Oracle Database のバージョンは 11.2.0.4 以降である必要があります。
- データベースと Kafka クラスターは同じリージョンに存在している必要があります。別のリージョンを使用する場合、追加のデータ転送料金が発生する可能性があることに注意してください。
- ネットワークに関する考慮事項については、「ネットワークアクセス」を参照してください。静的なエグレス IP を使用する方法については、「静的なエグレス IP アドレス」を参照してください。
- Confluent CLI がインストールされ、クラスター用に構成されていること。「Confluent CLI のインストール」を参照してください。
- スキーマレジストリ ベースのフォーマット(Avro、JSON_SR(JSON スキーマ)、Protobuf など)を使用するには、Schema Registry を有効にしておく必要があります。詳細については、「環境の制限」を参照してください。
- シンクコネクターを作成する前に、Confluent Cloud クラスター上にソース Kafka トピックが 1 つ以上存在している必要があります。
- 「データベースの考慮事項」も参照してください。
Confluent Cloud Console の使用¶
ステップ 1: Confluent Cloud クラスターを起動します。¶
インストール手順については、「Confluent Cloud を使用した Apache Kafka のクイックスタート」を参照してください。
ステップ 2: コネクターを追加します。¶
左のナビゲーションメニューの Data integration をクリックし、Connectors をクリックします。クラスター内に既にコネクターがある場合は、+ Add connector をクリックします。
Step 4: Enter the connector details.¶
注釈
- Ensure you have all your prerequisites completed.
- アスタリスク( * )は必須項目であることを示しています。
At the Add Oracle Database Sink Connector screen, complete the following:
If you've already populated your Kafka topics, select the topic(s) you want to connect from the Topics list.
To create a new topic, click +Add new topic.
- Select the way you want to provide Kafka Cluster credentials. You can
choose one of the following options:
- Global Access: Allows your connector to access everything you have access to. With global access, connector access will be linked to your account. This option is not recommended for production.
- Granular access: Limits the access for your connector. You will be able to manage connector access through a service account. This option is recommended for production.
- Use an existing API key: Allows you to enter an API key and secret part you have stored. You can enter an API key and secret (or generate these in the Cloud Console).
- Click Continue.
- Enter the following Databricks Delta Lake connection details:
- Connection host: The JDBC connection host.
- Database name: The JDBC database name.
- Connection port: The JDBC connection port.
- SSL mode: The SSL mode to use to connect to your database.
- Connection user: The JDBC connection user.
- Connection password: The JDBC connection password.
- Trust store: The trust store file that contains the server CA certificate.
- Distinguished name (DN) of the database server: Used to specify the distinguished name (DN) of the database server. Only required if using verify-full as the SSL mode.
- Trust store password: The password for the trust store file that contains the server CA certificate.
- Click Continue.
注釈
Configuration properties that are not shown in the Cloud Console use the default values. See 構成プロパティ for all property values and definitions.
Select the Input Kafka record value format (data coming from the Kafka topic): AVRO, JSON_SR, or PROTOBUF. A valid schema must be available in Schema Registry to use a schema-based message format.
Select an insert mode:
INSERT
: 標準的なINSERT
行関数を使用します。該当する行が既にテーブルに存在する場合は、エラーが発生します。UPSERT
: このモードはINSERT
と似ています。ただし、該当する行が既に存在する場合に、UPSERT
関数は、指定された値で列の値を上書きします。
Show advanced configurations
Auto create table: Whether to automatically create the destination table if it is missing.
Auto add columns: Whether to automatically add columns in the table if they are missing.
Database timezone: Name of the JDBC timezone that should be used in the connector when inserting time-based values.
Table name format: A format string for the destination table name, which may contain
${topic}
as a placeholder for the originating topic name. For example, to create a table namedkafka-orders
based on a Kafka topic namedorders
, you would enterkafka-${topic}
in this field.Table types: The comma-separated types of database tables to which the sink connector can write.
Fields included: List of comma-separated record value field names. If empty, all fields from the record value are used.
PK mode: The primary key mode.
PK Fields: List of comma-separated primary key field names.
When to quote SQL identifiers: When to quote table names, column names, and other identifiers in SQL statements.
Max rows per batch: Maximum number of rows to include in a single batch when polling for new data. This setting can be used to limit the amount of data buffered internally in the connector.
Input Kafka record key format: Sets the input Kafka record key format. Valid entries are AVRO, JSON_SR, PROTOBUF, STRING. A valid schema must be available in Schema Registry to use a schema-based message format.
Delete on null: Whether to treat null record values as deletes. Requires
pk.mode
to berecord_key
.For Transforms and Predicates, see the Single Message Transforms (SMT) documentation for details.
See 構成プロパティ for all property values and definitions.
Click Continue.
Based on the number of topic partitions you select, you will be provided with a recommended number of tasks.
- To change the number of recommended tasks, enter the number of tasks for the connector to use in the Tasks field.
- Click Continue.
Verify the connection details.
Click Launch.
コネクターのステータスが Provisioning から Running に変わります。
Step 5: Check for records.¶
行がデータベースに取り込まれていることを確認します。
Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。
ちなみに
コネクターを起動すると、デッドレターキューのトピックが自動的に作成されます。詳細については、「Confluent Cloud デッドレターキュー」を参照してください。
Confluent CLI の使用¶
以下の手順に従うと、Confluent CLI を使用してコネクターをセットアップし、実行できます。
注釈
- すべての 前提条件 を満たしていることを確認してください。
- コマンド例では Confluent CLI バージョン 2 を使用しています。詳細については、「Confluent CLI v2 への移行 <https://docs.confluent.io/confluent-cli/current/migrate.html#cli-migrate>`__」を参照してください。
ステップ 2: コネクターの必須の構成プロパティを表示します。¶
以下のコマンドを実行して、コネクターの必須プロパティを表示します。
confluent connect plugin describe <connector-catalog-name>
例:
confluent connect plugin describe OracleDatabaseSink
出力例:
Following are the required configs:
connector.class: OracleDatabaseSink
input.data.format
name
kafka.auth.mode
kafka.api.key
kafka.api.secret
connection.host
connection.port
connection.user
connection.password
db.name
ssl.server.cert.dn
ssl.rootcertfile
tasks.max
topics
ステップ 3: コネクターの構成ファイルを作成します。¶
コネクター構成プロパティを含む JSON ファイルを作成します。以下の例は、コネクターの必須プロパティを示しています。構成プロパティの値と詳細については、「構成プロパティ」を参照してください。
{
"connector.class": "OracleDatabaseSink",
"input.data.format": "AVRO",
"name": "OracleDatabaseSink_0",
"kafka.auth.mode": "KAFKA_API_KEY",
"kafka.api.key": "<my-kafka-api-key>",
"kafka.api.secret": "<my-kafka-api-secret>",
"connection.host ": "<connection-host",
"connection.port": "1521",
"connection.user": "<user-name>",
"connection.password": "<user-password>",
"db.name": "<database-name>",
"ssl.server.cert.dn": "<distinquished-database-server-name>",
"ssl.rootcertfile": "<certificate-text>",
"tasks.max": "1",
"topics": "<topic-name>",
}
以下のプロパティ定義に注意してください。
"connector.class"
: コネクターのプラグイン名を指定します。"input.data.format"
: Kafka 入力レコード値のフォーマット(Kafka トピックから送られるデータ)を設定します。指定可能なエントリは、AVRO、JSON_SR、および PROTOBUF です。スキーマベースのメッセージフォーマット(たとえば、Avro、JSON_SR(JSON スキーマ)、および Protobuf)を使用するには、Confluent Cloud Schema Registry を構成しておく必要があります。その他のオプションについては、「構成プロパティ」のinput.key.format
を参照してください。"name"
: 新しいコネクターの名前を設定します。
"kafka.auth.mode"
: 使用するコネクターの認証モードを指定します。オプションはSERVICE_ACCOUNT
またはKAFKA_API_KEY
(デフォルト)です。API キーとシークレットを使用するには、構成プロパティkafka.api.key
とkafka.api.secret
を構成例(前述)のように指定します。サービスアカウント を使用するには、プロパティkafka.service.account.id=<service-account-resource-ID>
に リソース ID を指定します。使用できるサービスアカウントのリソース ID のリストを表示するには、次のコマンドを使用します。confluent iam service-account list
例:
confluent iam service-account list Id | Resource ID | Name | Description +---------+-------------+-------------------+------------------- 123456 | sa-l1r23m | sa-1 | Service account 1 789101 | sa-l4d56p | sa-2 | Service account 2
"connection.<...>"
: データベース接続のプロパティ。connection.host
は、database-1.<id>.us-west-2.rds.amazonaws.com
のように入力します。詳細については、『Database Connection Details』を参照してください。"ssl.rootcertfile"
: デフォルトのssl.mode
はverify-full
です。プロパティssl.rootcertfile
を使用し、プロパティ値のテキスト証明書ファイルのコンテンツを追加します。たとえば、"ssl.rootcertfile": "<certificate-text>"
のようになります。その他のssl.mode
オプションについては、「構成プロパティ」を参照してください。"ssl.server.cert.dn"
: デフォルトのssl.mode
はverify-full
です。このモードでは、サーバーの識別名を指定する必要があります。その他のssl.mode
オプションについては、「構成プロパティ」を参照してください。"tasks.max"
: このコネクターで使用できる タスク の最大数を入力します。タスクが多いほどパフォーマンスが向上する可能性があります(複数のタスクを実行するとコンシューマーラグが減少します)。"topics"
: 特定のトピック名を指定するか、複数のトピック名をコンマ区切りにしたリストを指定します。
Single Message Transforms: CLI を使用する SMT の追加の詳細については、Single Message Transforms(SMT) のドキュメントを参照してください。
See 構成プロパティ for all property values and descriptions.
ステップ 4: プロパティファイルを読み込み、コネクターを作成します。¶
以下のコマンドを入力して、構成を読み込み、コネクターを起動します。
confluent connect create --config <file-name>.json
例:
confluent connect create --config oracle-db-sink-config.json
出力例:
Created connector OracleDatabaseSink_0 lcc-do6vzd
ステップ 5: コネクターのステータスを確認します。¶
以下のコマンドを入力して、コネクターのステータスを確認します。
confluent connect list
出力例:
ID | Name | Status | Type | Trace
+------------+--------------------------+---------+------+-------+
lcc-do6vzd | OracleDatabaseSink_0 | RUNNING | sink | |
ステップ 6: レコードを確認します。¶
行がデータベースに取り込まれていることを確認します。
Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。
ちなみに
コネクターを起動すると、デッドレターキューのトピックが自動的に作成されます。詳細については、「Confluent Cloud デッドレターキュー」を参照してください。
構成プロパティ¶
Use the following configuration properties with this connector.
Which topics do you want to get data from?¶
topics
Identifies the topic name or a comma-separated list of topic names.
- 型: list
- 重要度: 高
Input messages¶
input.data.format
Sets the input Kafka record value format. Valid entries are AVRO, JSON_SR, or PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.
- 型: string
- 重要度: 高
input.key.format
Sets the input Kafka record key format. This need to be set to a proper format if using pk.mode=record_key. Valid entries are AVRO, JSON_SR, PROTOBUF, STRING. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.
- 型: string
- 重要度: 高
delete.enabled
Whether to treat null record values as deletes. Requires pk.mode to be record_key.
- 型: boolean
- デフォルト: false
- 重要度: 低
How should we connect to your data?¶
name
Sets a name for your connector.
- 型: string
- Valid Values: A string at most 64 characters long
- 重要度: 高
Kafka Cluster credentials¶
kafka.auth.mode
Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode.
- 型: string
- Default: KAFKA_API_KEY
- Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY
- 重要度: 高
kafka.api.key
- 型: password
- 重要度: 高
kafka.service.account.id
The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
- 型: string
- 重要度: 高
kafka.api.secret
- 型: password
- 重要度: 高
How should we connect to your database?¶
connection.host
JDBC connection host.
- 型: string
- 重要度: 高
connection.port
JDBC connection port.
- 型: int
- Valid Values: [0,...,65535]
- 重要度: 高
connection.user
JDBC connection user.
- 型: string
- 重要度: 高
connection.password
JDBC 接続パスワード。
- 型: password
- 重要度: 高
db.name
JDBC データベース名。
- 型: string
- Valid Values: Must match the regex
^[a-zA-Z][a-zA-Z0-9$#_]*$
- 重要度: 高
ssl.mode
What SSL mode should we use to connect to your database. disabled disables SSL entirely. verify-ca uses SSL for encryption and performs authentication of the server CA. verify-ca option requires a Java truststore containing the server CA and the truststore password to be provided.
- 型: string
- デフォルト: verify-full
- 重要度: 高
ssl.truststorefile
The trust store containing server CA certificate. Only required if using verify-ca or verify-full ssl mode.
- 型: password
- Default: [hidden]
- 重要度: 低
ssl.truststorepassword
The trust store password containing server CA certificate. Only required if using verify-ca or verify-full ssl mode.
- 型: password
- Default: [hidden]
- 重要度: 低
ssl.server.cert.dn
Use this paramter to specify the distinguished name (DN) of the database server. Only required if using verify-full ssl mode.
- 型: string
- 重要度: 低
Database details¶
insert.mode
The insertion mode to use.
- 型: string
- デフォルト: INSERT
- 重要度: 高
table.name.format
A format string for the destination table name, which may contain ${topic} as a placeholder for the originating topic name.
For example, kafka_${topic} for the topic ‘orders’ will map to the table name ‘kafka_orders’.
- 型: string
- デフォルト: ${topic}
- 重要度: 中
table.types
The comma-separated types of database tables to which the sink connector can write. By default this is
TABLE
, but any combination ofTABLE
andVIEW
is allowed. Not all databases support writing to views, and when they do the sink connector will fail if the view definition does not match the records' schemas (regardless ofauto.evolve
).- 型: list
- デフォルト: TABLE
- 重要度: 低
fields.whitelist
List of comma-separated record value field names. If empty, all fields from the record value are utilized, otherwise used to filter to the desired fields.
- 型: list
- 重要度: 中
db.timezone
Name of the JDBC timezone used in the connector when querying with time-based criteria. Defaults to UTC.
- 型: string
- デフォルト: UTC
- 重要度: 中
Primary Key¶
pk.mode
The primary key mode, also refer to pk.fields documentation for interplay. Supported modes are:
none: No keys utilized.
kafka: Apache Kafka® coordinates are used as the PK.
record_value: Field(s) from the record value are used, which must be a struct.
record_key: Field(s) from the record key are used, which must be a struct.
- 型: string
- Valid Values: none, kafka, record_key, record_value
- 重要度: 高
pk.fields
List of comma-separated primary key field names. The runtime interpretation of this config depends on the pk.mode:
none: Ignored as no fields are used as primary key in this mode.kafka: Must be a trio representing the Kafka coordinates, defaults to __connect_topic,__connect_partition,__connect_offset if empty.
record_value: If empty, all fields from the value struct will be used, otherwise used to extract the desired fields.
- 型: list
- 重要度: 高
SQL/DDL Support¶
auto.create
Whether to automatically create the destination table if it is missing.
- 型: boolean
- デフォルト: false
- 重要度: 中
auto.evolve
Whether to automatically add columns in the table if they are missing.
- 型: boolean
- デフォルト: false
- 重要度: 中
quote.sql.identifiers
When to quote table names, column names, and other identifiers in SQL statements. For backward compatibility, the default is ‘always’.
- 型: string
- デフォルト: ALWAYS
- Valid Values: ALWAYS, NEVER
- 重要度: 中
Connection details¶
batch.sizes
新しいデータのポーリング時に単一のバッチに含める最大行数。この設定を使用して、コネクターの内部にバッファリングするデータの量を制限できます。
- 型: int
- デフォルト: 3000
- Valid Values: [1,...,5000]
- 重要度: 低
Number of tasks for this connector¶
tasks.max
- 型: int
- Valid Values: [1,...]
- 重要度: 高
データベースの考慮事項¶
以下の問題に留意します。
auto.create=true
の場合、string 型は CLOB にマッピングされます。たとえば、次のような Avro スキーマがあるとします。{ "connect.name": "ksql.ratings", "fields": [ { "name": "rating_id", "type": "long" }, { "name": "user_id", "type": "int" }, ... { "name": "channel", "type": "string" }, { "name": "message", "type": "string" } ], "name": "ratings", "namespace": "ksql", "type": "record" }
これらの値は、テーブルスキーマの CLOB にマッピングされます。
Name Null? Type ----------- -------- ---------- rating_id NOT NULL NUMBER(19) user_id NOT NULL NUMBER(10) stars NOT NULL NUMBER(10) route_id NOT NULL NUMBER(10) rating_time NOT NULL NUMBER(19) channel NOT NULL CLOB message NOT NULL CLOB
String は、
auto.create=true
の場合に CLOB にマッピングされるので、String 型を使用するフィールドをプライマリキーとして使用することはできません。String 型のフィールドをプライマリキーとして使用する場合は、まずデータベースにテーブルを作成し、その後、auto.create=false
を使用します。これを行わない場合、以下の行を含む例外が発生します。... "stringValue": "Exception chain:\njava.sql.SQLException: ORA-02329: column of datatype LOB cannot be unique or a primary key ...
テーブル名と列名では、大文字と小文字が区別されます。たとえば、次のような Avro スキーマがあるとします。
{ "connect.name": "ksql.pageviews", "fields": [ { "name": "viewtime", "type": "long" }, { "name": "userid", "type": "string" }, { "name": "pageid", "type": "string" } ], "name": "pageviews", "namespace": "ksql", "type": "record" }
PAGEVIEWS
という名前のテーブルが作成され、pageviews
が見つからないという例外が発生します。create table pageviews ( userid VARCHAR(10) NOT NULL PRIMARY KEY, pageid VARCHAR(50), viewtime VARCHAR(50) ); Table PAGEVIEWS created. DESC pageviews; Name Null? Type -------- -------- ------------ USERID NOT NULL VARCHAR2(10) PAGEID VARCHAR2(50) VIEWTIME VARCHAR2(50)
以下のような例外メッセージが DLQ に配置されます。
{ "key": "__connect.errors.exception.message", "stringValue": "Table \"pageviews\" is missing and auto-creation is disabled" }
この問題を解決するには、最初に Oracle Database にテーブルを作成し、
auto.create=false
を使用します。create table "pageviews" ( "userid" VARCHAR(10) NOT NULL PRIMARY KEY, "pageid" VARCHAR(50), "viewtime" VARCHAR(50) ); Table "pageviews" created. DESC "pageviews"; Name Null? Type -------- -------- ------------ userid NOT NULL VARCHAR2(10) pageid VARCHAR2(50) viewtime VARCHAR2(50)
注釈
SQL 標準では、識別子とキーワードについて、引用符で囲まれている場合を除き、大文字と小文字を区別しないようにデータベースを定義しています。つまり、
CREATE TABLE test_case
ではTEST_CASE
という名前のテーブルが作成され、CREATE TABLE "test_case"
ではtest_case
という名前のテーブルが作成されます。これは、テーブル列識別子の場合も同様です。識別子の引用符付けの詳細については、『Database Identifiers, Quoting, and Case Sensitivity』を参照してください。
次のステップ¶
参考
フルマネージド型の Confluent Cloud コネクターが Confluent Cloud ksqlDB でどのように動作するかを示す例については、「Cloud ETL のデモ」を参照してください。この例では、Confluent CLI を使用して Confluent Cloud のリソースを管理する方法についても説明しています。