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

Oracle Database Technology Night #71
Oracle Database 23c マイクロサービス関連新機能
Ryunosuke Deguchi
日本オラクル株式会社
クラウド・エンジニアリング統括
Autonomous & Analytics ソリューション部
2023/9/28

View Slide

1. Transactional Event Queues (TxEventQ)関連新機能
2. Oracle Database Sagas
3. Oracle SQL Access to Kafka
Agenda
Copyright © 2023, Oracle and/or its affiliates
2

View Slide

サービス間の影響を極小化しシステムの変更容易性を高めるアーキテクチャ
マイクロサービス・アーキテクチャとは
3
保守とテストの容易性
• 分割したサービス毎に組織を編成し開発・運用の自由度を高める
• 更新単位を最小限にすることでテスト規模を最小化
疎結合
• API化や非同期化によりサービス間の結合度を低減
• 変更による他の稼働中のサービスにへの影響を極小化
独立してデプロイ可能
• データソースやアプリケーション・モジュールをサービス毎で占有
• デプロイやスケールの変更の単位サービス毎で任意に最適化
API
サービス実装
データストア
イベント・ストア

View Slide

4
Transactional Event Queues
(TxEventQ)関連新機能

View Slide

Oracle Databaseに統合された堅牢で多機能なメッセージ・キューイング・システム
• TxEventQへのエンキュー/デキュー、TxEventQと他のメッセージングシステム間のメッセージ伝播機能を提供
• 標準のデータベース機能（リカバリ、セキュリティなど）がサポートされる
• データ駆動およびイベント駆動のアーキテクチャから求められる要件に対処
• あらゆる機能が自動化されたマネージドなサービス
• データベースと統合してパフォーマンスを最適化
Oracle Advanced Queuing (AQ) /
Oracle Transactional Event Queues (TxEventQ) とは
メッセージを送信する側のクライアントメッセージを消費する側のクライアント
Producer Consumer
エンキューデキュー
TxEventQ
伝播
伝播
TxEventQ TxEventQ
5

View Slide

Web サーバ
注文サービス
API/Brokers
モバイル
IoT
Producers Consumers
モバイルアプリ
検索ポータル
ダッシュボード
リアルタイム分析、
アラート
機械学習モデル
分析レポート
アドホック分析
Oracle Database
(コンバージド/
イベント・キュー)
トランザクション
• データの統合
• マイクロサービスのサポート
• オープンなインタフェース
• 簡単なアプリ構築とAPI
• バックアップ、セキュリティなど
の管理タスクの軽減
Oracle Advanced Queuing (AQ) /
Oracle Transactional Event Queues (TxEventQ) とは
6
OCI
オブジェクト・
ストア
OCI
アーカイブ・
ストア
データ + イベント

View Slide

Oracle TxEventQはOracle AQシャード・キューの21c以降の更新バージョン
Oracle AQおよびトランザクション・イベント・キュー(TxEventQ)の歴史
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
に名称変更

View Slide

AQ シャード・キューから進化した新しいメッセージ・キューイング・システム
• 伝播機能
• イベントストリームの同時実行性が向上
• TxEventQ用のKafka Javaクライアント/Kafka実装の拡張
• 多くの言語のサポート
• リアルタイムなパフォーマンス監視
• AQからTxEventQへのオンライン移行ツール
AQシャード・キューからTxEventQへ
Producer Consumer
TxEventQ
TxEventQ
8
TxEventQ 伝播伝播

View Slide

リモート・サブスクライバへの伝播
• メッセージは1つのキューから別のキューに伝播できる
• アプリケーションは同じデータベース・キューに接続されていなくても相互に通信可
（宛先キューは同じデータベースでもリモート・データベースでも可能）➡ リモートでの処理のオフロード、バックアップ
• 伝播により、多くの受信者にメッセージを展開可能、異なるキューのメッセージを1つのキューに結合することもできる
• JMSセッション・レベルの順序付けセマンティクスを利用し、宛先のキューに送信
Oracle AQ / TxEventQ のメッセージ伝播
P
伝播
伝播
C
伝播
異なるキューのメッセージを1つのキューに伝播
P
P
C
C
1つのキューから異なる複数の別のキューに伝播
9

View Slide

リモート・サブスクライバへの伝播
2種類の伝播
• キューからdblinkへの伝播
• キューからキューへの伝播
Oracle AQ / TxEventQ のメッセージ伝播
P
キューからdblinkへの伝播
10
dblink
サブスクライブ・キュー
キューからキューへの伝播
P
dblink,
destination_queue
宛先キュー

View Slide

TxEventQ
イベントストリームとは
イベントストリーム（シャード）：キュー表のパーティションで構成
• 水平分割による高い同時実行性とスループットを実現する
• イベントストリームは自動でパーティション化
• インスタンス内にイベントストリームでエンキューセッションを分散
• インスタンス内にすべてのEvent Streamsのデキュー
• ローカルデキューアがない場合にメッセージをクロスインスタンス転送
イベントストリームの同時実行性が向上
RACインスタンス2
キュー表
ProducerC
イベントストリーム1
キュー表
ProducerA
ProducerB
Consumer
Consumer
Consumer
11

View Slide

新機能：パーティション化されたイベントストリーム
• TEQではイベントストリームがパーティション化されるように
• 新しいパーティションは、必要に応じて自動的に作成される
• パーティション内のすべてのメッセージがデキューされると、パーティションは切り捨てられ再利用される
12
キュー表
インスタンス
Consumer
ProducerA
パーティション1
キュー表
a1 a2 a3
a1
a2
a3

View Slide

新機能：RACを使用したときの同時実行性が向上
• パーティション化されたイベントストリームとRACインスタンスにアフィニティを持たせる
• １つのイベントストリームの処理を１つのRACインスタンス内で完結させるようにする
• 元々RACはサポートしていたが、新たにサポートされたパーティショニングと組み合わせて実現
13
キュー表
イベントストリーム
ProducerA Consumer
パーティション
キュー表
イベントストリーム
ProducerA Consumer
パーティション
a1
a2
a1
a2

View Slide

Apache Kafkaとは？
• Apache Kafkaは、スケーラビリティに優れた分散メッセージキュー
• 広く使用され、人気のあるオープンソースのイベントストリーミングおよびメッセージングシステム
• 分散型のフォールトトレラントアーキテクチャで大量のデータを処理する機能を備える。
KafkaとOracle Databaseの互換性
Producer Consumer
Kafka Brokers
• Broker：Kafkaが稼働する 1 台のサーバー
• Kafka Cluster：Kafkaが実行されているサーバ(Broker)をグループ化したもの
• Zookeeper：Kafkaを管理するサーバ
14

View Slide

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のAPIを使用してTxEventQへのエンキュー・デキューが
可能に
KafkaとTxEventQのメッセージのやり取りが透過的に
Producer Consumer
Producer Consumer
Producer Consumer
TEQ
TEQ
TEQ
TEQ
Producer Consumer
API
15

View Slide

アーキテクチャ
プロデューサプロデューサプロデューサ
コンシューマコンシューマコンシューマ
プロデューサプロデューサプロデューサ
コンシューマコンシューマコンシューマ
Kafkaブローカ
props.put("bootstrap.servers", ":9092") props.put("bootstrap.servers", ":1521")
TxEventQ
Zookeeper
トピック1
KafkaトピックおよびパーティションはTxEventQキュー表およびイベント・ストリームにマップされる
プロデューサ1
プロデューサ2
コンシューマ1
コンシューマ2
コンシューマ3
相互運用
または
置換可能
0 1
0 1
0 1 2
16

View Slide

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 producer = new KafkaProducer(props);
Future lastFuture = null; int msgCnt = 20000;
for(int i=0;iProducerRecord producerRecord = new ProducerRecord("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と連動させる方法：
この行を確認します。
こちら
17

View Slide

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 producer = new KafkaProducer(props);
Future lastFuture = null; int msgCnt = 20000;
for(int i=0;iProducerRecord producerRecord = new ProducerRecord("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をブートストラップ・
サーバー・リストに追加
コードの残りの
部分の変更は
必要なし！
18

View Slide

•複数のコンシューマおよび遅
延送信を含む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
多くの言語のサポート
19

View Slide

TxEventQでモニタリング可能な要素
• TxEventQのパフォーマンス監視でモニタリングできる内容は次の通り：
メッセージング・システムの正常性の確認
エンキュー/デキューのスループット、キューの深さを含む、全体的な主要パフォーマンス指標の監視
メッセージング・アクティビティに由来するCPUの負荷、メモリーの使用状況、およびデータベース待機クラスの監視
各キューの正常性状態を確認して、パフォーマンスの低いキューを素早く特定
• TxEventQのパフォーマンス監視は、データベースの動的パフォーマンス・ビューから情報を取得
• キューの主要メトリックに関するレポートの出力は各種ユーザー・インタフェースと統合可能
• オープン・ソース・ツールPrometheusおよびGrafanaを使用してメトリックの監視が可能
リアルタイムなパフォーマンス監視
20

View Slide

TxEventQモニタ・システムの構成
• TxEventQのメトリックを視覚的に取得可能
Oracle Database
TxEventQ
Prometheus Grafana
Oracle Databaseから監視に
必要な情報を収集するための
エージェント
エクスポータを使用して収集
したメトリックを保管するサー
バー
Prometheusの情報を
使用してデータを視覚化
するプラットフォーム
Prometheus の
エクスポータ
(oracledb_exporter)
dockerコンテナ上で動作
TxEventQのメトリックを取
得する
データソースを
呼ぶ
Grafanaで視覚化されたメトリック
次のようなメトリック情報を取得：
 ステータス
 キューの数
 サブスクライバの数
 エンキュー/デキューのスループット
 メッセージの数
21

View Slide

Grafanaの使用によりサマリーを視覚的に確認することが可能に
TxEventQモニターではインスタンス・キュー・サブスクライバ・ディスク・グループを対象に次のサマリーを確認することが可
能：
• すべてのTxEventQ全体のサマリー
• データベース・メトリックのサマリー
• システム・メトリックのサマリー
• TxEventQごとのサブスクライバのサマリー
▲ システム・サマリー・ダッシュボード：
システム・レベルのメトリックとキュー・レベルのメトリックを表示
（CPU使用率と使用メモリーなど）
▲ TxEventQサマリー・ダッシュボード：
全体的に集計されたTxEventQの統計情報(ステータス、キューの
数、サブスクライバの数、エンキュー率/デキュー率、メッセージの数
など)
▶ データベース・サマリー・ダッシュボード：
全体的なDBのパフォーマンスと統計情報
22

View Slide

• 移行ステップを自動化するPL/SQLパッケージDBMS_AQMIGTOOLの提供
• 機能
• AQの定義とデータをチェックし、移行の許可/不許可、適応可能かどうかのレポートと推奨事項の表示
• 要件に応じた移行モードが選択可能：AUTOMATIC/INTERACTIVE/OFFLINE/ONLY_DEFINITION
• 移行のコミット/フォールバックが選択可能
• キューの移行履歴の確認
AQ から TxEventQ へのオンライン移行ツール
23
TxEventQ
との互換性
チェック
Start
CHECK_MIGRATION_
TO_TXEVENTQ
INT_MIGRATION
移行実行
End
ABORT_MIGRATION
COMMIT_MIGRATION
非サポート機能なし
リスト生成→非サポート機能リスト
成功

View Slide

AQ シャード・キューから進化した新しいメッセージ・キューイング・システム
• 伝播機能
• イベントストリームの同時実行性が向上
• TxEventQ用のKafka Javaクライアント/Kafka実装の拡張
• 多くの言語のサポート
• リアルタイムなパフォーマンス監視
• AQからTxEventQへのオンライン移行ツール
AQシャード・キューからTxEventQへ
Producer Consumer
TxEventQ
TxEventQ
24
TxEventQ 伝播伝播

View Slide

Oracle Database Sagas
25
データ整合性を備えた分散トランザクション処理の実装

View Slide

1. Sagaパターンとは
2. Oracle DatabaseのSaga実装
Agenda
26

View Slide

Sagaパターンとは
27
サービス間をまたがるトランザクションの結果整合性の担保

View Slide

結果整合性による一貫性
相手のサービスのデータソースに直接アクセスできない
• トランザクションによる一貫性が利用できない
• 自サービスと相手のサービスそれぞれの「結果整合性」により全体の一貫性を保つ
マイクロサービスにおける一貫性の考え方
28
在庫管理
サービス
注文管理
サービス
他のサービスのデータソース
に直接アクセスできない

View Slide

サービス間をまたがるトランザクションの仕組み
Sagaパターン
補償トランザクションによる(事後)結果整合性
• 処理の成功を前提とした楽観的な呼出し手法
• 一部の処理が失敗した場合に、既に完了した処理を
取り消す「補償トランザクション」により整合性を取る
TCC (Try/Confirm/Cancel) パターン
予約ベースの(事前)結果整合性
• 処理完了の可否を事前に確認した上で処理を実施
• Tryフェーズにより不整合の生じる処理を行わないこと
で整合性を担保
マイクロサービスで検討される代表的な一貫性の仕組み
29
Insert
Delete
Insert
Reserve
Update
Reserve
Update

View Slide

サービス間をまたがるトランザクションの仕組み
補償トランザクションによる(事後)結果整合性
• 処理の成功を前提とした楽観的な呼出し手法
• 一部の処理が失敗した場合に、すでに完了した処理を取り消す「補償トランザクション」により整合性を取る
向いているトランザクション制御の特性
• トランザクションの存続期間が長く、1 つのマイクロサービスが長時間実行されても、他のマイクロサービスがブロックされる
ことは避けたい
• ワークフロー内の操作が失敗した場合にロールバックできる必要がある
• 他のトランザクションと分離されやすく、補償処理に人手が関与しやすい
ユースケース
• ユーザー単位で行われる受付業務
• カート→チェックアウト→配送のような長期的なトランザクション
Sagaパターンとは
30
Insert
Delete
Insert

View Slide

Oracle DatabaseのSaga実装
31
データベース機能を利用したOSaga

View Slide

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

View Slide

Oracle Databaseを使用したSagaの実装(正常パターン)
33
Travel
Agency
Hotel
Service
Airline
Service
Saga参加者
Saga参加者
TravelAgencyPDB
BrokerPDB
HotelPDB
AirlinePDB
room表
flight表
1. ホテル予約リクエスト
2. ホテル予約処理
3. ホテル予約OK
4. 航空券予約リクエスト
5. 航空券予約処理
6. 航空券予約OK
7. 予約確定

View Slide

Oracle Databaseを使用したSagaの実装(異常パターン)
34
Travel
Agency
Hotel
Service
Airline
Service
Saga参加者
Saga参加者
TravelAgencyPDB
BrokerPDB
HotelPDB
AirlinePDB
room表
flight表

View Slide

35
Travel
Agency
Hotel
Service
Airline
Service
Saga参加者
Saga参加者
TravelAgencyPDB
BrokerPDB
HotelPDB
AirlinePDB
room表
flight表
5. 航空券予約処理失敗
6. 航空券予約NG

View Slide

36
Travel
Agency
Hotel
Service
Airline
Service
Saga参加者
Saga参加者
TravelAgencyPDB
BrokerPDB
HotelPDB
AirlinePDB
room表
flight表
5. 航空券予約処理失敗
6. 航空券予約NG
7. ロールバック指示
8. ロールバック
(自動補正)

View Slide

優れたデータ一貫性の提供
• Saga実装はデータベースに統合
• Sagaのコード化、デプロイ、メンテンナンスが容易
• Saga中断時のロールバック処理を行う組み込み済の自動補正ロジック(ロック・フリー列値の予約機能)
• Oracle Databaseのスケーラビリティ、高可用性、一貫性、堅牢性の活用
• Oracle Multitenantによる一元的なDB管理
• ディクショナリ表やビューですべてのSaga参加者の一元的な状態管理が可能
• アプリ開発者のハードル低減
• 障害時のリカバリ・ロジックの記述が不要
• 組み込み済のAQおよびTxEventQにより、メッセージおよびイベントの生成が可能－Kafkaコーディネータ不要
• より高レベルのAPIおよびSpring Bootとの統合
• Javaアノテーションの使用－既存コードの流用が可能
• コンバージド・データベースの活用
• アプリケーション・ペイロードのJSONサポート
Oracle Databaseを使用したSagaの実装
37

View Slide

23c Oracle Saga Frameworkを試してみたその1：設定編
23c Oracle Saga Frameworkを試してみたその2：実行編
参考：Oracle Databaseを使用したSagaの実装サンプル(PL/SQL)
38

View Slide

Oracle SQL Access to Kafka
39

View Slide

Oracle SQL APIを使用してKafkaトピックに動的にクエリを実行できる
• Oracle Database 23cからのネイティブ機能
• SQL構文でKafkaストリーミングデータを処理
• Oracle DatabaseにKafkaアクセスが統合されているため、外部クライアントが不要
• DBMS_KAFKA および DBMS_KAFKA_ADMパッケージで構成される
Oracle SQL access to Kafka(OSaK)とは
Producer
0 1 2
0 1
0
Topic
Partition0
Partition1
Partition2
40

View Slide

KafkaデータとOracle Database内のデータを組み合わせたデータ分析
• データベースに保存せずに一時的な利用が可能
データ処理はOracle Databaseのトランザクションとして実行・制御
• トランザクションはデータベースのACID(原子性・一貫性・独立性・永続性）要件に準拠
• トランザクションでオフセットを管理することで、システム障害時のKafkaレコードの消失やアプリケーションによる再処理を
防ぐ（Kafkaデータの分離と耐久性の向上）
アプリ開発者でなくともストリーミング・データの活用が可能
• データ処理はSQL, PL/SQLで記述
Oracle SQL access to Kafka(OSaK)のメリット
Oracle SQL APIを使用してKafkaトピックに動的にクエリを実行
41

View Slide

3つのデータ・アクセス・モード
• ロード：Kafkaトピックのデータをデータベースのテーブルにロードし、様々なアプリからアクセス可能にする。主に
DWH用途。
• ストリーミング：Kafkaレコードを順番に1度だけSQLやPL/SQLを使って処理する。
• シーカブル：指定した開始と終了のタイムスタンプ間のKafkaレコードにアクセスする。過去の時点のデータをまとめ
て取得する用途。
OSaKがサポートしているデータ形式
• 区切りテキストデータ(csvなど)
• JSON
• Avro
Oracle SQL access to Kafka(OSaK) の機能
42

View Slide

• ロード：Kafkaトピックのデータをデータベースのテーブルにロードし、様々なアプリからアクセス可能にする。主に
DWH用途。
43
0 1 2
0 1
0
Topic
Partition0
Partition1
Partition2
リアルタイム分析
機械学習モデル
分析レポート

View Slide

• ストリーミング：Kafkaレコードを順番に1度だけSQLやPL/SQLを使って処理する。
44
0 1
0
Topic
Partition0
Partition1
Partition2
0 1 2
ユーザー表
0 User1
1 User2

View Slide

• シーカブル：指定した開始と終了のタイムスタンプ間の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
・・・

View Slide

1. 各Kafkaクラスタのクラスタ・アクセス・ディレクトリの作成
• ディレクトリのREAD権限を与えることで、どのOracle DatabaseユーザーがKafkaクラスタにアクセスできるか制
御するのに使用
2. ORACLE_BASEのディレクトリにクラスタ・アクセス設定ファイルを格納するOSディレクトリの作成
3. SYSDBAでデータベースにログインし、ディレクトリ・オブジェクトを作成
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

View Slide

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プロパティを使用している場合）
security.protocol=SASL_SSL #Kafkaブローカーとの通信に使用されるセキュリティ・プロトコル
sasl.mechanism=PLAIN #認証に使用するSASLメカニズム
sasl.username=// #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

View Slide

6. 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;
--登録成功の場合0が出力される
DBMS_KAFKA_ADM.REGISTER_CLUSTER('KAFKACLUS1','CELL-1.STREAMING.AP-TOKYO-1.OCI.OR
--------------------------------------------------------------------------------
0
48

View Slide

1. ロードする表の作成
2. ロード・アプリケーションの作成
例：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

View Slide

3. メッセージのPublish
4. ロード・アプリケーションの実行
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

View Slide

5. ロードしたデータの確認
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

View Slide

View Slide

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

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

oracle4engineer
PRO

More Decks by oracle4engineer

Other Decks in Technology

Featured

Transcript

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

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

oracle4engineer PRO

More Decks by oracle4engineer

Other Decks in Technology

Featured

Transcript

oracle4engineer
PRO