Upgrade to Pro — share decks privately, control downloads, hide ads and more …

TechNight#71 - Oracle Database 23c 新機能#1 Micros...

TechNight#71 - Oracle Database 23c 新機能#1 Microservices関連新機能

TechNight#71 Oracle Database 23c 新機能#1 Microservice関連新機能

oracle4engineer

September 29, 2023
Tweet

More Decks by oracle4engineer

Other Decks in Technology

Transcript

  1. Oracle Database Technology Night #71 Oracle Database 23c マイクロサービス関連新機能 Ryunosuke

    Deguchi 日本オラクル株式会社 クラウド・エンジニアリング統括 Autonomous & Analytics ソリューション部 2023/9/28
  2. 1. Transactional Event Queues (TxEventQ)関連新機能 2. Oracle Database Sagas 3.

    Oracle SQL Access to Kafka Agenda Copyright © 2023, Oracle and/or its affiliates 2
  3. サービス間の影響を極小化しシステムの変更容易性を高めるアーキテクチャ マイクロサービス・アーキテクチャとは Copyright © 2023, Oracle and/or its affiliates 3

    保守とテストの容易性 • 分割したサービス毎に組織を編成し開発・運用の自由度を高める • 更新単位を最小限にすることでテスト規模を最小化 疎結合 • API化や非同期化によりサービス間の結合度を低減 • 変更による他の稼働中のサービスにへの影響を極小化 独立してデプロイ可能 • データソースやアプリケーション・モジュールをサービス毎で占有 • デプロイやスケールの変更の単位サービス毎で任意に最適化 API サービス実装 データストア イベント・ストア
  4. Oracle Databaseに統合された堅牢で多機能なメッセージ・キューイング・システム • TxEventQへのエンキュー/デキュー、TxEventQと他のメッセージングシステム間のメッセージ伝播機能を提供 • 標準のデータベース機能(リカバリ、セキュリティなど)がサポートされる • データ駆動およびイベント駆動のアーキテクチャから求められる要件に対処 • あらゆる機能が自動化されたマネージドなサービス

    • データベースと統合してパフォーマンスを最適化 Oracle Advanced Queuing (AQ) / Oracle Transactional Event Queues (TxEventQ) とは メッセージを送信する側のクライアント メッセージを消費する側のクライアント Producer Consumer エンキュー デキュー TxEventQ 伝播 伝播 TxEventQ TxEventQ Copyright © 2023, Oracle and/or its affiliates 5
  5. Web サーバ 注文サービス API/Brokers モバイル IoT Producers Consumers モバイルアプリ 検索ポータル

    ダッシュボード リアルタイム分析、 アラート 機械学習モデル 分析レポート アドホック分析 Oracle Database (コンバージド/ イベント・キュー) トランザクション • データの統合 • マイクロサービスのサポート • オープンなインタフェース • 簡単なアプリ構築とAPI • バックアップ、セキュリティなど の管理タスクの軽減 Oracle Advanced Queuing (AQ) / Oracle Transactional Event Queues (TxEventQ) とは 6 OCI オブジェクト・ ストア OCI アーカイブ・ ストア データ + イベント Copyright © 2023, Oracle and/or its affiliates
  6. Oracle TxEventQはOracle AQシャード・キューの21c以降の更新バージョン Oracle AQおよびトランザクション・イベント・キュー(TxEventQ)の歴史 Copyright © 2023, Oracle and/or

    its affiliates 7 Oracle8 Database Oracle8i Database Oracle Database 11g Oracle9i Database Oracle9i Database Release2 Oracle Database 12c, 18c, 19c Oracle Database 21c, 23c TxEventQ AQ クラシック・キュー オブジェク/ADT; IOT; パーティション化 AQ シングル・コ ンシューマキュー AQマルチコン シューマキュー Oracle Real Application Clusters AQ JMS 標準サポート AQ パフォーマンス 最適化 Oracle Streams AQ に名称変更 Transactional Event Queues Kafka互換性の追加 (高性能メッセージング / Events DB) Oracle AQ Oracle AQ 単一データベース・ワー クロード AQ シャード・ キュー AQ JMS シャード・ キュー Oracle AQ に名称変更
  7. AQ シャード・キューから進化した新しいメッセージ・キューイング・システム • 伝播機能 • イベントストリームの同時実行性が向上 • TxEventQ用のKafka Javaクライアント/Kafka実装の拡張 •

    多くの言語のサポート • リアルタイムなパフォーマンス監視 • AQからTxEventQへのオンライン移行ツール AQシャード・キューからTxEventQへ Producer Consumer TxEventQ TxEventQ Copyright © 2023, Oracle and/or its affiliates 8 TxEventQ 伝播 伝播 エンキュー デキュー
  8. リモート・サブスクライバへの伝播 2種類の伝播 • キューからdblinkへの伝播 • キューからキューへの伝播 Oracle AQ / TxEventQ

    のメッセージ伝播 P キューからdblinkへの伝播 Copyright © 2023, Oracle and/or its affiliates 10 dblink サブスクライブ・キュー キューからキューへの伝播 P dblink, destination_queue 宛先キュー
  9. TxEventQ イベントストリームとは イベントストリーム(シャード):キュー表のパーティションで構成 • 水平分割による高い同時実行性とスループットを実現する • イベントストリームは自動でパーティション化 • インスタンス内にイベントストリームでエンキューセッションを分散 •

    インスタンス内にすべてのEvent Streamsのデキュー • ローカルデキューアがない場合にメッセージをクロスインスタンス転送 イベントストリームの同時実行性が向上 RACインスタンス2 キュー表 RACインスタンス1 ProducerC イベントストリーム1 イベントストリーム2 キュー表 イベントストリーム1 ProducerA ProducerB Consumer Consumer Consumer Copyright © 2023, Oracle and/or its affiliates 11
  10. Apache Kafkaとは? • Apache Kafkaは、スケーラビリティに優れた分散メッセージキュー • 広く使用され、人気のあるオープンソースのイベントストリーミングおよびメッセージングシステム • 分散型のフォールトトレラントアーキテクチャで大量のデータを処理する機能を備える。 KafkaとOracle

    Databaseの互換性 Producer Consumer Kafka Brokers • Broker:Kafkaが稼働する 1 台のサーバー • Kafka Cluster:Kafkaが実行されているサーバ(Broker)をグループ化したもの • Zookeeper:Kafkaを管理するサーバ Copyright © 2023, Oracle and/or its affiliates 14
  11. TxEventQがKafkaのインターフェースOKafkaを持つように • Kafka APIアプリケーションとOracle APIアプリケーショ ンがメッセージの互換性を持つ • Kafka Java APIはOracle

    Databaseサーバーに接続し、メッ セージング・プラットフォームとしてTxEventQを使用可能 • KafkaをTxEventQに置き換えての使用が可能 • KafkaとTxEventQが相互にメッセージのやりとりできる ように • Kafka Java APIを使用することでKafkaからTxEventQへ、 TxEventQからKafkaへ相互にメッセージを送れるように KafkaとOracle Databaseの互換性 KafkaのAPIを使用してTxEventQへのエンキュー・デキューが 可能に KafkaとTxEventQのメッセージのやり取りが透過的に Producer Consumer Producer Consumer Producer Consumer TEQ TEQ TEQ TEQ Producer Consumer API Copyright © 2023, Oracle and/or its affiliates 15
  12. アーキテクチャ KafkaとOracle Databaseの互換性 プロデューサ プロデューサ プロデューサ コンシューマ コンシューマ コンシューマ プロデューサ

    プロデューサ プロデューサ コンシューマ コンシューマ コンシューマ Kafkaブローカ props.put("bootstrap.servers", "<kafka>:9092") props.put("bootstrap.servers", "<db>:1521") TxEventQ Zookeeper トピック1 KafkaトピックおよびパーティションはTxEventQキュー表およびイベント・ストリームにマップされる プロデューサ1 プロデューサ2 パーティション1 コンシューマ1 コンシューマ2 コンシューマ3 相互運用 または 置換可能 パーティション3 0 1 0 1 パーティション2 0 1 2 Copyright © 2023, Oracle and/or its affiliates 16
  13. import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import java.util.Properties; public class SimpleProducerOKafka {

    public static void main(String[] args) { try { Properties props = new Properties(); props.put("bootstrap.servers", “kafka:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); Producer<String, String> producer = new KafkaProducer<String, String>(props); Future<RecordMetadata> lastFuture = null; int msgCnt = 20000; for(int i=0;i<msgCnt;i++) { ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("TxEventQ", i+"", "Test message # " +i); lastFuture = producer.send(producerRecord); } System.out.println("Produced "+ msgCnt +" messages."); lastFuture.get(); producer.close(); } catch(Exception e) { System.out.println("Exception in Main " + e ); e.printStackTrace(); } } } Kafka Javaクライアント-プロデューサ 既存のKafkaクライアント・アプリケーションを TxEventQと連動させる方法: この行を確認します。 こちら Copyright © 2023, Oracle and/or its affiliates 17
  14. import org.oracle.okafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import java.util.Properties; public class SimpleProducerOKafka {

    public static void main(String[] args) { try { Properties props = new Properties(); props.put("bootstrap.servers", "database:1521"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); Producer<String, String> producer = new KafkaProducer<String, String>(props); Future<RecordMetadata> lastFuture = null; int msgCnt = 20000; for(int i=0;i<msgCnt;i++) { ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("TxEventQ", i+"", "Test message # " +i); lastFuture = producer.send(producerRecord); } System.out.println("Produced "+ msgCnt +" messages."); lastFuture.get(); producer.close(); } catch(Exception e) { System.out.println("Exception in Main " + e ); e.printStackTrace(); } } } TxEventQのKafka Javaクライアント-プロデューサ このインポートをTxEventQ Kafka プロデューサに変更 Oracle Databaseをブートストラップ・ サーバー・リストに追加 コードの残りの 部分の変更は 必要なし! Copyright © 2023, Oracle and/or its affiliates 18
  15. •複数のコンシューマおよび遅 延送信を含むJMS P2Pおよ びpub/sub •Spring JMSサポート用の Spring Bootスタータ •Kafkaブローカをメッセージ・ブ ローカとしてTxEventQに置き

    換えるためのKafka Javaクラ イアントのサポート(Okafka) Java JavaScript (Node.js) Python C/C++ ODP.NET PL/SQL RESTおよび SQLcl // get a connection to the database oracledb.initOracleClient({}); connection = await oracledb.getConnection({ user: 'pdbadmin', password: '********', connectString: 'db:1521/pdb1' }) // enqueue a message const rawQueue = await connection.getQueue("my_TxEventQ"); await rawQueue.enqOne(“Hi Mom!"); await connection.commit(); import oracledb from os import environ as env oracledb.init_oracle_client() connection = oracledb.connect(dsn='172.17.0.2:1521/pdb1',user='mark' ,password='******') queue = connection.queue("myq", "JSON") queue.enqOne(connection.msgproperties(payload={"develop er day": "today"})) connection.commit() Python Node.js 多くの言語のサポート Copyright © 2023, Oracle and/or its affiliates 19
  16. TxEventQモニタ・システムの構成 リアルタイムなパフォーマンス監視 • TxEventQのメトリックを視覚的に取得可能 Oracle Database TxEventQ Prometheus Grafana Oracle

    Databaseから監視に 必要な情報を収集するための エージェント エクスポータを使用して収集 したメトリックを保管するサー バー Prometheusの情報を 使用してデータを視覚化 するプラットフォーム Prometheus の エクスポータ (oracledb_exporter) dockerコンテナ上で動作 TxEventQのメトリックを取 得する データソースを 呼ぶ Grafanaで視覚化されたメトリック 次のようなメトリック情報を取得:  ステータス  キューの数  サブスクライバの数  エンキュー/デキューのスループット  メッセージの数 Copyright © 2023, Oracle and/or its affiliates 21
  17. Grafanaの使用によりサマリーを視覚的に確認することが可能に TxEventQモニターではインスタンス・キュー・サブスクライバ・ディスク・グループを対象に次のサマリーを確認することが可 能: • すべてのTxEventQ全体のサマリー • データベース・メトリックのサマリー • システム・メトリックのサマリー •

    TxEventQごとのサブスクライバのサマリー リアルタイムなパフォーマンス監視 ▲ システム・サマリー・ダッシュボード: システム・レベルのメトリックとキュー・レベルのメトリックを表示 (CPU使用率と使用メモリーなど) ▲ TxEventQサマリー・ダッシュボード: 全体的に集計されたTxEventQの統計情報(ステータス、キューの 数、サブスクライバの数、エンキュー率/デキュー率、メッセージの数 など) ▶ データベース・サマリー・ダッシュボード: 全体的なDBのパフォーマンスと統計情報 Copyright © 2023, Oracle and/or its affiliates 22
  18. • 移行ステップを自動化するPL/SQLパッケージDBMS_AQMIGTOOLの提供 • 機能 • AQの定義とデータをチェックし、移行の許可/不許可、適応可能かどうかのレポートと推奨事項の表示 • 要件に応じた移行モードが選択可能:AUTOMATIC/INTERACTIVE/OFFLINE/ONLY_DEFINITION • 移行のコミット/フォールバックが選択可能

    • キューの移行履歴の確認 AQ から TxEventQ へのオンライン移行ツール Copyright © 2023, Oracle and/or its affiliates 23 TxEventQ との互換性 チェック Start CHECK_MIGRATION_ TO_TXEVENTQ INT_MIGRATION 移行実行 End ABORT_MIGRATION COMMIT_MIGRATION 非サポート機能なし リスト生成→非サポート機能リスト 成功
  19. AQ シャード・キューから進化した新しいメッセージ・キューイング・システム • 伝播機能 • イベントストリームの同時実行性が向上 • TxEventQ用のKafka Javaクライアント/Kafka実装の拡張 •

    多くの言語のサポート • リアルタイムなパフォーマンス監視 • AQからTxEventQへのオンライン移行ツール AQシャード・キューからTxEventQへ Producer Consumer TxEventQ TxEventQ Copyright © 2023, Oracle and/or its affiliates 24 TxEventQ 伝播 伝播 エンキュー デキュー
  20. Oracle Database Sagas Copyright © 2023, Oracle and/or its affiliates

    25 データ整合性を備えた分散トランザクション処理の実装
  21. サービス間をまたがるトランザクションの仕組み Sagaパターン 補償トランザクションによる(事後)結果整合性 • 処理の成功を前提とした楽観的な呼出し手法 • 一部の処理が失敗した場合に、既に完了した処理を 取り消す「補償トランザクション」により整合性を取る TCC (Try/Confirm/Cancel)

    パターン 予約ベースの(事前)結果整合性 • 処理完了の可否を事前に確認した上で処理を実施 • Tryフェーズにより不整合の生じる処理を行わないこと で整合性を担保 マイクロサービスで検討される代表的な一貫性の仕組み Copyright © 2023, Oracle and/or its affiliates 29 Insert Delete Insert Reserve Update Reserve Update
  22. サービス間をまたがるトランザクションの仕組み 補償トランザクションによる(事後)結果整合性 • 処理の成功を前提とした楽観的な呼出し手法 • 一部の処理が失敗した場合に、すでに完了した処理を取り消す「補償トランザクション」により整合性を取る 向いているトランザクション制御の特性 • トランザクションの存続期間が長く、1 つのマイクロサービスが長時間実行されても、他のマイクロサービスがブロックされる

    ことは避けたい • ワークフロー内の操作が失敗した場合にロールバックできる必要がある • 他のトランザクションと分離されやすく、補償処理に人手が関与しやすい ユースケース • ユーザー単位で行われる受付業務 • カート→チェックアウト→配送のような長期的なトランザクション Sagaパターンとは Copyright © 2023, Oracle and/or its affiliates 30 Insert Delete Insert
  23. 組み込みコーディネータによるシンプルで堅牢性の高いアーキテクチャ Oracle Databaseを使用したSagaの実装 Copyright © 2023, Oracle and/or its affiliates

    32 Travel Agency Hotel Service Airline Service メッセージ・ブローカー 旅行代理店サービス Sagaイニシエータ/コーディネータ ホテル予約サービス Saga参加者 航空券予約サービス Saga参加者 TravelAgencyPDB BrokerPDB HotelPDB AirlinePDB DBLINKを使用したDB間 のメッセージ伝播 AQ/TxEventQによ る複数の参加者間の 非同期通信 SagaID SagaID JSONサポート SagaID 補償トランザクション の自動実行
  24. 組み込みコーディネータによるシンプルで堅牢性の高いアーキテクチャ Oracle Databaseを使用したSagaの実装(正常パターン) Copyright © 2023, Oracle and/or its affiliates

    33 Travel Agency Hotel Service Airline Service メッセージ・ブローカー 旅行代理店サービス Sagaイニシエータ/コーディネータ ホテル予約サービス Saga参加者 航空券予約サービス Saga参加者 TravelAgencyPDB BrokerPDB HotelPDB AirlinePDB room表 flight表 1. ホテル予約リクエスト 2. ホテル予約処理 3. ホテル予約OK 4. 航空券予約リクエスト 5. 航空券予約処理 6. 航空券予約OK 7. 予約確定
  25. 組み込みコーディネータによるシンプルで堅牢性の高いアーキテクチャ Oracle Databaseを使用したSagaの実装(異常パターン) Copyright © 2023, Oracle and/or its affiliates

    34 Travel Agency Hotel Service Airline Service メッセージ・ブローカー 旅行代理店サービス Sagaイニシエータ/コーディネータ ホテル予約サービス Saga参加者 航空券予約サービス Saga参加者 TravelAgencyPDB BrokerPDB HotelPDB AirlinePDB room表 flight表 1. ホテル予約リクエスト 2. ホテル予約処理 3. ホテル予約OK 4. 航空券予約リクエスト
  26. 組み込みコーディネータによるシンプルで堅牢性の高いアーキテクチャ Oracle Databaseを使用したSagaの実装(異常パターン) Copyright © 2023, Oracle and/or its affiliates

    35 Travel Agency Hotel Service Airline Service メッセージ・ブローカー 旅行代理店サービス Sagaイニシエータ/コーディネータ ホテル予約サービス Saga参加者 航空券予約サービス Saga参加者 TravelAgencyPDB BrokerPDB HotelPDB AirlinePDB room表 flight表 1. ホテル予約リクエスト 2. ホテル予約処理 3. ホテル予約OK 4. 航空券予約リクエスト 5. 航空券予約処理失敗 6. 航空券予約NG
  27. 組み込みコーディネータによるシンプルで堅牢性の高いアーキテクチャ Oracle Databaseを使用したSagaの実装(異常パターン) Copyright © 2023, Oracle and/or its affiliates

    36 Travel Agency Hotel Service Airline Service メッセージ・ブローカー 旅行代理店サービス Sagaイニシエータ/コーディネータ ホテル予約サービス Saga参加者 航空券予約サービス Saga参加者 TravelAgencyPDB BrokerPDB HotelPDB AirlinePDB room表 flight表 5. 航空券予約処理失敗 6. 航空券予約NG 7. ロールバック指示 8. ロールバック (自動補正)
  28. 優れたデータ一貫性の提供 • Saga実装はデータベースに統合 • Sagaのコード化、デプロイ、メンテンナンスが容易 • Saga中断時のロールバック処理を行う組み込み済の自動補正ロジック(ロック・フリー列値の予約機能) • Oracle Databaseのスケーラビリティ、高可用性、一貫性、堅牢性の活用

    • Oracle Multitenantによる一元的なDB管理 • ディクショナリ表やビューですべてのSaga参加者の一元的な状態管理が可能 • アプリ開発者のハードル低減 • 障害時のリカバリ・ロジックの記述が不要 • 組み込み済のAQおよびTxEventQにより、メッセージおよびイベントの生成が可能-Kafkaコーディネータ不要 • より高レベルのAPIおよびSpring Bootとの統合 • Javaアノテーションの使用-既存コードの流用が可能 • コンバージド・データベースの活用 • アプリケーション・ペイロードのJSONサポート Oracle Databaseを使用したSagaの実装 Copyright © 2023, Oracle and/or its affiliates 37
  29. 23c Oracle Saga Frameworkを試してみた その1:設定編 23c Oracle Saga Frameworkを試してみた その2:実行編

    参考:Oracle Databaseを使用したSagaの実装サンプル(PL/SQL) Copyright © 2023, Oracle and/or its affiliates 38
  30. Oracle SQL APIを使用してKafkaトピックに動的にクエリを実行できる • Oracle Database 23cからのネイティブ機能 • SQL構文でKafkaストリーミングデータを処理 •

    Oracle DatabaseにKafkaアクセスが統合されているため、外部クライアントが不要 • DBMS_KAFKA および DBMS_KAFKA_ADMパッケージで構成される Copyright © 2023, Oracle and/or its affiliates Oracle SQL access to Kafka(OSaK)とは Producer 0 1 2 0 1 0 Topic Partition0 Partition1 Partition2 40
  31. KafkaデータとOracle Database内のデータを組み合わせたデータ分析 • データベースに保存せずに一時的な利用が可能 データ処理はOracle Databaseのトランザクションとして実行・制御 • トランザクションはデータベースのACID(原子性・一貫性・独立性・永続性)要件に準拠 • トランザクションでオフセットを管理することで、システム障害時のKafkaレコードの消失やアプリケーションによる再処理を

    防ぐ(Kafkaデータの分離と耐久性の向上) アプリ開発者でなくともストリーミング・データの活用が可能 • データ処理はSQL, PL/SQLで記述 Copyright © 2023, Oracle and/or its affiliates Oracle SQL access to Kafka(OSaK)のメリット Oracle SQL APIを使用してKafkaトピックに動的にクエリを実行 41
  32. 3つのデータ・アクセス・モード • ロード:Kafkaトピックのデータをデータベースのテーブルにロードし、様々なアプリからアクセス可能にする。主に DWH用途。 Copyright © 2023, Oracle and/or its

    affiliates Oracle SQL access to Kafka(OSaK) の機能 Oracle SQL APIを使用してKafkaトピックに動的にクエリを実行 43 0 1 2 0 1 0 Topic Partition0 Partition1 Partition2 リアルタイム分析 機械学習モデル 分析レポート
  33. 3つのデータ・アクセス・モード • ストリーミング:Kafkaレコードを順番に1度だけSQLやPL/SQLを使って処理する。 Copyright © 2023, Oracle and/or its affiliates

    Oracle SQL access to Kafka(OSaK) の機能 Oracle SQL APIを使用してKafkaトピックに動的にクエリを実行 44 0 1 0 Topic Partition0 Partition1 Partition2 0 1 2 ユーザー表 0 User1 1 User2
  34. 3つのデータ・アクセス・モード • シーカブル:指定した開始と終了のタイムスタンプ間のKafkaレコードにアクセスする。過去の時点のデータをまとめ て取得する用途。 Copyright © 2023, Oracle and/or its

    affiliates Oracle SQL access to Kafka(OSaK) の機能 Oracle SQL APIを使用してKafkaトピックに動的にクエリを実行 45 Topic Partition0 Partition1 Partition2 0 1 2 3 4 5 6 7 8 ・・・ 0 1 2 3 ・・・ 0 1 ・・・ T1~T6 T1 2 T2 3 T3 4 ・・・
  35. 1. 各Kafkaクラスタのクラスタ・アクセス・ディレクトリの作成 • ディレクトリのREAD権限を与えることで、どのOracle DatabaseユーザーがKafkaクラスタにアクセスできるか制 御するのに使用 2. ORACLE_BASEのディレクトリにクラスタ・アクセス設定ファイルを格納するOSディレクトリの作成 3. SYSDBAでデータベースにログインし、ディレクトリ・オブジェクトを作成

    Copyright © 2023, Oracle and/or its affiliates Oracle SQL access to Kafka(OSaK) Kafkaクラスタの登録手順 SQL> create directory osakaccess_kafkaclus1 AS ‘’; mkdir –p /u01/app/oracle/osak/kafkaclus1/config; SQL> create directory OSAK_KAFKACLUS1_CONFIG AS ‘/u01/app/oracle/osak/kafkaclus1/config’; 46
  36. 4. Kafkaクラスタにアクセスするために必要な情報を含む設定ファイルosakafka.propertiesの作成 • OSaK管理者によって作成され、管理される • キーと値の組(key=value)のテキストファイル • $ORACLE_BASE/osak/cluster-name/configに格納される 以下はOCI Streaming

    Service(OSS) Kafkaクラスタにアクセスするためのosakafka.properties設定例 5. クレデンシャルの作成(SSL.key.locationまたはsasl.usernameプロパティを使用している場合) Copyright © 2023, Oracle and/or its affiliates Oracle SQL access to Kafka(OSaK) Kafkaクラスタの登録手順 security.protocol=SASL_SSL #Kafkaブローカーとの通信に使用されるセキュリティ・プロトコル sasl.mechanism=PLAIN #認証に使用するSASLメカニズム sasl.username=<tenancyName>/<username>/<streamPoolID> #Kafkaクラスタとの認証に必要なユーザー名 max.partition.fetch.bytes=1048576 #1パーティションにつき1MBまでのリクエストサイズ制限(推奨) SQL> BEGIN dbms_credential.create_credential( credential_name => ‘KAFKACLUS1CRED1’, username => ‘KAFKACLUS1’, password => ‘ssl-key-password or sasl-password’) END; / 47
  37. 6. SYSDBAとしてログインし、DBMS_KAFKA_ADM.REGISTER_CLUSTERプロシージャでKafkaクラスタを登録 Copyright © 2023, Oracle and/or its affiliates Oracle

    SQL access to Kafka(OSaK) 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; --登録成功の場合0が出力される DBMS_KAFKA_ADM.REGISTER_CLUSTER('KAFKACLUS1','CELL-1.STREAMING.AP-TOKYO-1.OCI.OR -------------------------------------------------------------------------------- 0 48
  38. 1. ロードする表の作成 2. ロード・アプリケーションの作成 Copyright © 2023, Oracle and/or its

    affiliates Oracle SQL access to Kafka(OSaK) 例:KafkaデータをOracle Databaseのテーブルにロード SQL> create table address( name varchar2(30), address varchar2(100), city varchar2(30), state varchar2(2), zipcode integer); SQL> DECLARE v_options varchar2(50); BEGIN v_options := ‘{“fmt”:”DSV”, “reftable”:”address”}’; DBMS_KAFKA.CREATE_LOAD_APP(‘KAFKACLUS1’, ‘AddressApp’, ‘OSaKtestStream’, v_options); END; / 49
  39. 3. メッセージのPublish 4. ロード・アプリケーションの実行 Copyright © 2023, Oracle and/or its

    affiliates Oracle SQL access to Kafka(OSaK) 例:KafkaデータをOracle Databaseのテーブルにロード SQL> DECLARE v_records_inserted integer; BEGIN v_records_inserted := 2; DBMS_KAFKA.EXECUTE_LOAD_APP(‘KAFKACLUS1’, ‘AddressApp’, ‘address’, v_records_inserted); END; / $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 50
  40. 5. ロードしたデータの確認 Copyright © 2023, Oracle and/or its affiliates Oracle

    SQL access to Kafka(OSaK) 例:KafkaデータをOracle Databaseのテーブルにロード 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 51