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 をクリックします。

ステップ 3: コネクターを選択します。¶

Oracle Database Sink コネクターのアイコンをクリックします。

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.

注釈

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 named kafka-orders based on a Kafka topic named orders, you would enter kafka-${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 be record_key.
- For Transforms and Predicates, see the Single Message Transforms (SMT) documentation for details.
See 構成プロパティ for all property values and definitions.
Click Continue.

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>`__」を参照してください。

ステップ 1: 使用可能なコネクターをリスト表示します。¶

以下のコマンドを入力して、使用可能なコネクターをリスト表示します。

confluent connect plugin list

ステップ 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 of TABLE and VIEW 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 of auto.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 のリソースを管理する方法についても説明しています。