[OCI]Oracle Streaming Service with Kafka Connect (2020/08/25)

Oracle Streaming Service with Kafka Connect (2020/08/25)

https://blogs.oracle.com/developers/oracle-streaming-service-with-kafka-connect
投稿者：Igor Aragao de Souza | Principal Big Data Consultant


このブログ記事は、Apache Kafka ConnectでのOracle Streaming Service (OSS)の使用法について説明する入門的なブログ記事です。

Oracle Streaming Serviceの入門編については、以下のブログ記事をご覧ください。



たくさんのコードを作るのではなく、Kafka Connect を使って、設定ファイルだけでたくさんの接続ができることを知っていましたか？
JDBC, HDFS, Object Storeなどと簡単に接続することができます。

Kafka Connect (または Connect API) は、他のシステムとの間でデータをインポート/エクスポートするためのフレームワークです。
コネクタは外部システムに接続するためのシンプルな方法を提供することを目的としており、必要なのは設定ファイルだけです。
JDBC、Object Store、HDFS、Elasticsearchなどの一般的なもののためのコネクタはConfluent Hubにすでに存在しています。
公式ブログの発表と概要

ソース コネクタを使用すると、外部ソースからデータを取り込むことができます。
シンクコネクタを使用すると、データを外部ソースに配信することができます。


私の概要のブログ記事をチェックしてください。
Kafka Connect の完全な概要や詳細については Kafka のドキュメントを参照してください。

Confluentは、あなたが見て、利用可能なすべての接続をダウンロードすることができますページを提供しています。

Oracle Streaming ServiceはKafkaと互換性があり、
OSSをKafkaコネクトで使用することができ、
すべての単語の最高のものを得ることができます。
これでJDBC、Object Store、HDFS、Elasticsearchなどとの接続がコンフィグファイルを変更するだけの本当に簡単な方法でできるようになりました。

Kafka Connect Harness - Oracle Streaming Serviceが発表した
Kafkaと同社のフルマネージド有料ストリーミング・サービスとの互換性により、幅広いユースケースが切り開かれます。


Oracle Streaming Service + Kafka Connect Harnessを使用することで、
開発者はコードをリファクタリングすることなく、完全に管理されたサービスに移行することができます。

Kafka Connect Harnessは、Kafka Connect用の一部のメタデータを保存し、OSSが他のKafkaコネクタと接続できるようにするOSSの一部です。
次の図は、Kafka Connectのセットアップの全体的なアーキテクチャを示しています。

左側のSourceとSinkはデータの元となるもので、ファイルシステム、データベース、オブジェクトストアなど何でも構いません。
この部分はオンプレミスでもどこでも構いませんが、右は OSS と Kafka Connect Harness が入っている OCI の部分です。
真ん中はKafkaのソースとシンクの部分を追加したところで、これはホストしてデプロイする方法を確認する必要がある部分です。

アーキテクチャの概要


ここでは、Kafka Connect のデプロイ方法について、いくつかのオプションを紹介します。

デプロイのオプション


Kafka ConnectでOSSを利用するには、いくつかの方法があります。

  1. オンプレミスのKafka
  2. Docker
  3. OKE
  4. OCI

これは網羅的なリストではありません。

ここでは、いくつかのオプションを示すブログ記事が素敵です。

スタート地点


1) オンプレミスのKafka

JAR ファイルをダウンロードし、Kafka Connect ワーカーのフォルダに配置します。

Kafka Connect ワーカーの設定ファイル *.properties を探し、エディタで開きます。
plugin.path 設定を検索し、コネクタが存在するフォルダを含むように修正または作成します。

Kafka Connect 設定ファイルの connect-standalone.properties (または connect-distributed.properties) は、プラグインを参照します。

plugin.path=/opt/connector/plugin


考慮事項:これは簡単でシンプルなソリューションで、始めるには良いですが、大きなスループットにはスケールしません。
テストや開発環境に適しています。

2) Docker

ConfluentはここにDockerイメージを提供していますが、DebeziumのようなKafka connectのDockerイメージを使うこともできます。

dockerイメージの中にjarをコピーするだけですが、どのイメージを使っているかによってパスが異なります。

Debeziumの例

docker run -it --rm --name weather-connect-demo -p 8083:8083 -e GROUP_ID=1
\ -e BOOTSTRAP_SERVERS="bootstrap_URL"
\ -e CONFIG_STORAGE_TOPIC=”ID-config”
\ -e OFFSET_STORAGE_TOPIC=”ID-offset”
\ -e STATUS_STORAGE_TOPIC=”ID-status”
\ -v path/to/local/your_connect.jar/:/kafka/connect/your_connect
\ debezium/connect:latest


考慮事項：設定と起動が簡単で、すべて設定済みのイメージを使用できる。

3) OKE（Oracle Kubernetes Engine

Kubernetes上でKafkaを実行することができます。

https://www.confluent.io/blog/getting-started-apache-kafka-kubernetes/

https://medium.com/@vamsiramakrishnan/ingest-iot-data-from-mqtt-brokers-into-oci-oracle-streaming-service-oci-kafka-connect-harness-and-301c5b8a9be7

Strimziでも実行できます。

Oracle Kubernetes Engine(OKE)上のStrimzi + Oracle Streaming Service(OSS)上のKafka Connect

考慮事項: 回復力、高可用性、および耐障害性は、水平方向に拡張されます。
しかし、設定や起動が困難なため、些細なユースケースでは多すぎると考える人もいるかもしれませんが、追加のオーバーヘッドをもたらします。

4) OCI

ここでの考え方は、オンプレミスのKafkaを使うのと同じですが、Kafka自体をOCIでホスティングします。
もちろん、ここではdockerイメージを実行したり、Kubernetesを実行したりすることもできます。

考慮事項: 設定と起動が簡単です。

Getting Started

まず、あなたの借家契約に適切なポリシーを追加する必要があります。

Allow group xyz to manage connect-harnesses in tenancy


認証


Kafka プロトコルで認証するには、auth-token を使う必要があります。
auth-tokenは、ユーザー詳細ページのコンソールで生成することができます。

専用のグループ/ユーザーを作成し、そのグループに右のコンパートメント(またはテナント全体)の
Connect Harnessを管理する権限を与えることを強くお勧めします。
作成したユーザーのauth-Tokenを生成し、Kafka connectの設定で使用することができます。

tenancyName/username/compartmentId


bootstrapサーバの場合は、リージョンのドメインに9092ポートを使用します。

streaming.{region}.oci.oraclecloud.com:9092


セットアップ


Kafka Connect をセットアップするには、Connect Harnessの OCID が必要です。

Connect Harnessは Java, Go, Python, Ruby SDK を使って作成するか、コンソールを使って作成します。



Connect Harnessを作成する簡単なJavaの例

CreateConnectHarnessDetails createConnectHarnessDetails = CreateConnectHarnessDetails.builder()
.compartmentId(compartment) //compartment where you want to create connect harness
.name("myConnectHarness") //connect harness name
.build();

CreateConnectHarnessRequest connectHarnessRequest = CreateConnectHarnessRequest.builder()
.createConnectHarnessDetails(createConnectHarnessDetails)
.build();

CreateConnectHarnessResponse createConnectHarnessResponse = streamAdminClient.createConnectHarness(connectHarnessRequest);
ConnectHarness connectHarness = createConnectHarnessResponse.getConnectHarness();

while (connectHarness.getLifecycleState() != ConnectHarness.LifecycleState.Active && connectHarness.getLifecycleState() != ConnectHarness.LifecycleState.Failed) {
GetConnectHarnessRequest getConnectHarnessRequest = GetConnectHarnessRequest.builder().connectHarnessId(connectHarness.getId()).build();
connectHarness = streamAdminClient.getConnectHarness(getConnectHarnessRequest).getConnectHarness();
}


List Connect Harnessの例

ListConnectHarnessesRequest listConnectHarnessesRequest = ListConnectHarnessesRequest.builder()
.compartmentId(comaprtment) // compartment id to list all the connect harnesses.
.lifecycleState(ConnectHarnessSummary.LifecycleState.Active)
.build();

ListConnectHarnessesResponse listConnectHarnessesResponse = streamAdminClient.listConnectHarnesses(listConnectHarnessesRequest);
List<ConnectHarnessSummary> items = listConnectHarnessesResponse.getItems();


Kafka Connect の設定




Kafka Connectには以下のプロパティが必要です。

bootstrap.servers=streaming.{region}.oci.oraclecloud.com:9092 security.protocol=SASL_PLAINTEXT sasl.mechanical=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="tenancyName/username/streamPoolID" password="authToken"




プロデューサーの設定


producer.security.protocol=SASL_PLAINTEXT
producer.sasl.mechanism=PLAIN
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="tenancyName/username/streamPoolID" password="authToken"


コンシューマーの設定


consumer.security.protocol=SASL_PLAINTEXT
consumer.sasl.mechanism=PLAIN
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="tenancyName/username/streamPoolID" password="authToken"


設定、ステータス、オフセットのトピック


offset.storage.topic=<connect-harness ocid>-offset
config.storage.topic=<connect-harness ocid>-config
status.storage.topic=<connect-harness ocid>-status


スロットル/リミット


各メッセージは最大100kbのサイズを持つことができます。リクエストスループットは50kb/sに調整されています。

また、最低でも200msのmaxWait時間を強制しています。

すべてのConnect Harnessは、最大で10,000個の鍵サイズ、または鍵の合計サイズが10mbを超えないように公開するように設定されています。

鍵の総数や鍵のサイズの合計が設定された上限に達すると、どのワーカーもメッセージを発行することができません。

最大制限に達した場合は、Kafka Connect Harness を再作成し、既存のものを削除することができる。

Photo by John Barkiple on Unsplash

コメント

コメントを投稿

このブログの人気の投稿

Oracle Database 19cサポート・タイムラインの重要な更新 (2024/11/20)

イメージ
Oracle Database 19cサポート・タイムラインの重要な更新 (2024/11/20) https://mikedietrichde.com/2024/11/20/important-update-to-oracle-database-19c-support-timelines/ 長い沈黙は、私が隠れているという意味ではありません。つまり、今はとても忙しいということです。そして、より少ない旅行で、私はあなたのコメントにもブログと返信するより多くの時間を持っていることを願っています。それでも、Oracle Database 19cサポート・タイムラインの非常に重要な更新について説明します。 * 更新された内容 2024年11月19日現在、Oracle Database 19cのサポート・タイムラインが調整されています。 MOSノート: 742060.1 – 現在のデータベース・リリースのリリース・スケジュールを見ると、新しい日付が表示されます。 Premier Support、2029年12月31日まで 2030年1月1日から2032年12月31日までの拡張サポート 2033年1月1日から2034年12月31日までのアップグレード・サポート これは知ることがとても大切です。 他のリリースの更新はありますか。 MOSノートの終わりにある変更ログ: 742060.1 – 現在のデータベース・リリースのリリース・スケジュールでは、Oracle Database 21cの最新の更新も確認できます。このリリースのPremier Support期間は、2027年7月31日まで延長されました。 また、Oracle Database 23aiのパッチ適用終了日は2032年12月31日に設定されています。 Lifetime Support Policy 更新は、Oracle Technology Productsの Oracle Lifetime Support Policy (6ページ)でも確認できます。 From: Oracle Lifetime Support Policy for Oracle Technology Product s – page 6 – Nov 19, 2024 特に、次の表に示す制限事項に注意してください。 Oracle Database 1...
続きを読む

ミリ秒の問題: BCCグループとOCIが市場データ・パフォーマンスを再定義する方法(AWSに対するベンチマークを使用) (2025/11/13)

イメージ
ミリ秒の問題: BCCグループとOCIが市場データ・パフォーマンスを再定義する方法(AWSに対するベンチマークを使用) (2025/11/13) https://blogs.oracle.com/cloud-infrastructure/post/milliseconds-matter-bcc-group-and-oci-vs-aws 投稿者： Steven Riley | Senior Cloud Architect William Bierds | Business Analyst and Systems Engineer - BCC Group International Travis Liles | Master Principal Cloud Architect Puru Patel | Cloud Architect - FSI これは、BCC グループと Oracle Cloud Infrastructure (OCI) 間の市場データ パートナーシップを紹介するブログ シリーズの第 2 部です。   はじめに 金融市場データアプリケーションは、トレーダー、金融機関、そして市場参加者が情報に基づいた意思決定を行えるよう、リアルタイムデータを提供することで、取引エコシステムにおいて重要な役割を果たしています。 前回のブログ で述べたように、取引環境において最適なパフォーマンスを確保するには、市場データを超低レイテンシかつ高い信頼性で配信する必要があります。 今日の急速に変化する金融市場において、クラウドは市場データ・アプリケーションにとって強力な推進力として台頭しています。適切なネットワークとインフラストラクチャがあれば、企業は実行時間の短縮、可用性の向上、そして堅牢なセキュリティを実現できます。しかし、企業によるクラウド導入は遅れており、ワークロードのクラウド移行計画を中止しているケースもあります。そこで、BCCグループとOracle Cloud Infrastructure（OCI）のパートナーシップが注目されています。 BCCグループは、主力市場データアプリケーションであるONE PlatformをOCI上に導入しました。このブログでは、以下の点について説明します。 このパートナーシップがなぜこれほど相乗効果をもたらすのかを探る OC...
続きを読む

Oracle SQLマクロ表: パラメータのあるビュー! (2025/05/01)

イメージ
Oracle SQLマクロ表: パラメータのあるビュー! (2025/05/01) https://medium.com/oracledevs/oracle-sql-macro-tables-finally-views-with-parameters-bc78286ab010 投稿者： vijay balebail View with parameters Oracleでは、強力でエレガントな機能であるSQLマクロが導入されました。この記事では、インライン・ビューのように動作するがスマートなパラメータ化されたSQLブロックを作成できるSQLマクロ表のみに焦点を当てます。 ビューにパラメータを渡したいと思ったことがあるなら、それはあなたの解決策です。 📌SQLマクロ表は「パラメータ付きビュー」と考えてください。 通常のSQLと同様に解析および最適化された問合せが返されます。ただし、複雑なPL/SQL構造に依存することなく、引数を渡すことができます。 SQLマクロテーブルで解決する問題 次のことを行います。 動的入力(検索語やしきい値など)の埋込み SQLロジックでの入力の使用(フィルタリングや順序付けなど) 多数の問合せまたはツールでロジックを再利用 ビューを本能的に使用しようとする場合もありますが、ビューはパラメータを使用できません。回避策は次のとおりです。 パイプラインPL/SQLファンクションの記述 カスタムSQLタイプの定義(オブジェクトと表) ループとPIPE ROWでロジックをラップ しかし、これはオプティマイザーフレンドリーではありません。 SQLマクロテーブルは、これをクリーンに解決します。 例: Top-N Vector類似性検索(Oracle 23ai) OracleのVECTOR_EMBEDDING関数を使用して質問を埋め込み、ベクトル・ストアから最も類似した上位N件の結果を取得するとします。 ベクトル検索を行うSQL問合せの例を次に示します。このクエリは、sqlplus から実行できます。vector_embdding関数を使用して質問を埋め込み、vector_distanceを使用してベクトル検索を実行することに注意してください。 WITH a AS ( SELECT VECTOR_EMBEDDING(ALL_MINILM...
続きを読む