Oracle SQL Access to Kafkaを使ってKafkaレコードをOracle Databaseのテーブルにロードする

# はじめに
Oracle SQL Access to Kafka(OSaK)は、SQLでKafkaトピックに動的にクエリを実行できる<a href="https://www.oracle.com/database/free/" target="_blank" rel="noopener noreferrer">Oracle Database 23ai Free</a>の新機能です。

概要やユースケースはこちらを参照ください。

https://qiita.com/wahagon/items/82429785f40adc5c418b

Oracle SQL Access to Kafkaには主にロード、ストリーミング、シーカブルといった3つの機能があります。
本記事では、その中でもKafkaレコードをOracle Databaseのテーブルにロードし、様々なアプリからアクセス可能にする**ローディング**機能の使用手順を紹介します。

# 前提条件
+ Oracle Database 23ai環境
+ Kafkaクラスタを作成していること
本記事ではOracle Cloud Infrastructure(OCI) 上で提供されるフルマネージドな分散メッセージングサービスである、OCI Streaming Service(OSS)を使ってKafkaクラスタを作成しています。
OSSについては、<a href="https://oracle-japan.github.io/ocitutorials/datascience/streaming-for-beginner/" target="_blank" rel="noopener noreferrer">OCI Streaming を動かしてみよう</a>というチュートリアルが参考になります。

# Oracle SQL Access to Kafka管理者の作成
まずは、Oracle SQL Access to Kafkaを管理するユーザ`KAFKA_ADMIN`を作成しておきます。
```sql
create user kafka_admin identified by <password>;
grant osak_admin_role to kafka_admin;
grant create session, create table, create view to kafka_admin;
```
`OSAK_ADMIN_ROLE`というロールを付与することで、そのユーザーが、Kafkaクラスターを構成、登録、および管理するために必要な以下のシステム権限が付与されます。
+ CREATE CREDENTIAL：Kafka SASL-SSL (Simple Authentication and Security Layer) パスワードまたは OSS (OCI Streaming Service) authToken を作成する権限
+ CREATE ANY DIRECTORY：クラスタアクセスおよびクラスタ構成ディレクトリを作成する権限
+ DROP ANY DIRECTORY：クラスタアクセスとクラスタ構成ディレクトリを削除する権限
+ DBMS_KAFKA_CLUSTERSに対するREAD権限
+ DBMS_KAFKA_APPLICATIONSに対するREAD権限
+ DBMS_KAFKA_MESSAGESに対するREAD権限

# Kafkaクラスタの登録
Kafkaクラスタに安全にアクセスするためには、osakafka.propertiesという構成ファイルと、Kerberos、SSL、PEMファイルによるTLS/SSLなどの選択したセキュリティ方式のセキュリティファイルと証明書が必要です。
これらはデータベース・ディレクトリ・オブジェクトに格納されます。

1. 空のパスを持つデータベース・ディレクトリ・オブジェクトを作成します。このディレクトリのREAD権限を与えることで、どのOracle DatabaseユーザがKafkaクラスタにアクセスできるか制御するのに使用します。今回は、クラスタ名はkafkaclus1とします。
    ```sql
    create directory osakaccess_kafkaclus1 AS '';
    ```
1. ORACLE_BASEのパスに、$ORACLE_BASE/osak/cluster_name/config（cluster_nameはKafkaクラスター名）というOSディレクトリを作成します。
    ```sh
    mkdir -p $ORACLE_BASE/osak/kafkaclus1/config;
    ```

1. SYSDBAとしてデータベースにログインし、対応するディレクトリ・オブジェクトを作成します。
※私の環境では$ORACLE_BASE=/opt/oracleです。
    ```sql
    create directory OSAK_KAFKACLUS1_CONFIG AS '/opt/oracle/osak/kafkaclus1/config';
    ```

1. Kafkaクラスタにアクセスするために必要な情報を記述する設定ファイルosakafka.propertiesを作成します。以下の特徴があります。
    + Oracle SQL Access to Kafka管理者によって作成され、管理される
    + キーと値の組(key=value)のテキストファイル
    + $ORACLE_BASE/osak/cluster-name/configに格納する

    以下はOCI Streaming Service(OSS)Kafkaクラスタにアクセスするためのosakafka.properties設定例です。
    その他のセキュリティ・プロトコル(SSLやKerberosなど)の設定例は、<a href="https://docs.oracle.com/en/database/oracle/oracle-database/23/sutil/oracle-sql-access-to-kafka.html#GUID-DB3FFB5D-B09A-4F17-86BA-48AA3BF0432A" target="_blank" rel="noopener noreferrer">ドキュメント</a>を参照してください。
    ```osakafka.properties
    security.protocol=SASL_SSL #Kafkaブローカーとの通信に使用されるセキュリティ・プロトコル
    sasl.mechanism=PLAIN #認証に使用するSASLメカニズム
    sasl.username=<tenancyName>/<username>/<streamPoolID> #Kafkaクラスタとの認証に必要なユーザー名
    max.partition.fetch.bytes=1048576 #1パーティションにつき1MBまでのリクエストサイズ制限(推奨)
    ```

1. osakafka.propertiesファイルで、SSL.key.locationまたはsasl.usernameのいずれかを使用している場合：
SSL SASL認証を使用するKafkaクラスタとの認証に必要なパスワードを格納するデータベースクレデンシャルを作成します。
    ```sql
    BEGIN
       dbms_credential.create_credential(
    	 credential_name => 'KAFKACLUS1CRED1',
         username => 'KAFKACLUS1',
         password => 'ssl-key-password or sasl-password')
    END;
    /
    ```

1. SYSDBAとしてログインし、DBMS_KAFKA_ADM.REGISTER_CLUSTER()プロシージャを使用してKafkaクラスタを登録します。
    ```sql
    select DBMS_KAFKA_ADM.REGISTER_CLUSTER('KAFKACLUS1', 'cell-1.streaming.ap-tokyo-1.oci.oraclecloud.com:9092', 'OSS', 'OSAKACCESS_KAFKACLUS1', 'KAFKACLUS1CRED1','OSAK_KAFKACLUS1_CONFIG', 'My test cluster kafkaclus1') from dual;
    ```
    各パラメータについては、<a href="https://docs.oracle.com/en/database/oracle/oracle-database/23/arpls/DBMS_KAFKA_ADM.html#GUID-0037A121-622A-4C60-9E6A-0DD9CC8C16FC" target="_blank" rel="noopener noreferrer">ドキュメント</a>を参照してください。

1. 登録成功の場合、以下のように0が出力されます。
    ```
    DBMS_KAFKA_ADM.REGISTER_CLUSTER('KAFKACLUS1','CELL-1.STREAMING.AP-TOKYO-1.OC
    ----------------------------------------------------------------------------
                                                                               0
    ```

# KafkaデータをOracle Databaseのテーブルにロード
Kafkaクラスタの登録が出来たので、続いてKafkaデータをテーブルにロードしてみます。

1. ロードする表addressを作成しておきます。
    ```sql
    create table address(
	    name varchar2(30),
	    address varchar2(100),
	    city varchar2(30),
	    state varchar2(2),
	    zipcode integer);
    ```

1. ロード・アプリケーションを作成します。Kafkaデータの形式が区切りテキストデータ(dsv)なので、fmtはdsvです。
    ```sql
    DECLARE
    v_options varchar2(100);
    BEGIN
       v_options := '{"fmt":"DSV", "reftable":"address"}';
       DBMS_KAFKA.CREATE_LOAD_APP('KAFKACLUS1', 'AddressApp', 'OSaKtestStream', v_options);
    END;
    /
    ```
    DBMS_KAFKA.CREATE_LOAD_APP(cluster-name, application-name, topic-name, v_options)とパラメータを指定しています。各パラメータについては、<a href="https://docs.oracle.com/en/database/oracle/oracle-database/23/arpls/DBMS_KAFKA.html#GUID-B4758885-4EA8-4943-8131-D1E0D680EFE1" target="_blank" rel="noopener noreferrer">ドキュメント</a>を参照してください。

1. OCI Streaming Service側で、メッセージ(住所データ)のPublishをしておきます。
    ```sh
    $KAFKA_HOME/bin/kafka-console-producer.sh \
    --bootstrap-server cell-1.streaming.ap-Tokyo-1.oci.oraclecloud.com:9092 \
    --topic OSaKtestStream \
    --producer.config $KAFKA_HOME/config/producer.properties

    >Fred Flintstone,2 BamBam Way,Bedrock,PA,04040
    >Barney Rubble,2 Dino Dr,Bedrock,PA,04040
    ```

1. ロード・アプリケーションを実行します。
    ```sql
    DECLARE
    v_records_inserted integer;
    BEGIN
       v_records_inserted := 2;
       DBMS_KAFKA.EXECUTE_LOAD_APP('KAFKACLUS1', 'AddressApp', 'address', v_records_inserted);
    END;
    /
    ```
    各パラメータについては、<a href="https://docs.oracle.com/en/database/oracle/oracle-database/23/arpls/DBMS_KAFKA.html#GUID-B3B138B1-2DAC-4D19-88B5-EBE4B1776D11" target="_blank" rel="noopener noreferrer">ドキュメント</a>を参照してください。

1. ロードしたデータの確認をしてみます。
    ```sql
    SELECT * FROM ADDRESS;
    
   NAME              ADDRESS              CITY            ST    ZIPCODE
   ---------------   ------------------   --------------  --    -----------
   Fred Flintstone   2 Bambam Way         Bedrock         PA    4040
   Barney Rubble     2 Dino Dr            Bedrock         PA    4040
   ```

   正しくロードされていることが確認できました。

このように、Oracle SQL Access to Kafkaを使えば、Kafkaデータを簡単にOracle Databaseのテーブルにロードすることができます。Kafkaデータに永続性を持たせたい場合や、DB内のデータと結合してデータ分析を行う場合に非常に便利です。

# 参考情報
<a href="https://docs.oracle.com/en/database/oracle/oracle-database/23/sutil/oracle-sql-access-to-kafka.html#GUID-EA66FD41-16B6-4E3F-A5ED-8C1897F1988F" target="_blank" rel="noopener noreferrer">Oracle SQL Access to Kafka</a>