問題タブ [ksqldb]

ksql cli を使用して confluent-4.1.1 に取り組んでいます。このバージョンでネストされた avro データ形式のストリームを作成できますか? バージョン 5 で試してみましたが、正常に動作していましたが、confluent-4.1.1 のネストされた avro スキーマについて言及されている場所に役立つリンクが見つかりませんでした

ストリームを作成しようとしました - ストリームを作成 new_order with(kafka_topic='transition',value_format='avro');

エラーが発生しました -

スキーマ レジストリから AVRO スキーマを取得できませんでした。avro タイプの正しいタイプが見つかりません: transition.Value

MySqlからKsqlへのデータ パイプラインを構築しようとしています。

ユース ケース: データ ソースは MySql です。MySqlでテーブルを作成しました。

私は使っている

スタンドアロン コネクタを開始します。そして、それはうまく機能しています。

トピック名でコンシューマーを開始しています。

MySQLテーブルにデータを挿入すると、コンシューマーでも結果が得られます。同じトピック名で KSQL Stream を作成しました。Kstreamでも同じ結果が期待されていますが、実行しているときに結果が得られません

コネクタ構成 -- source-quickstart-mysql.properties

サンプルデータ

• MySql

1.) データベースの作成:

2.) データベースを使用する:

3.) テーブルを作成します。

4.) テーブルにデータを挿入します。

• KSQL

1.) ストリームを作成します。

2.) クエリを選択します。

期待される出力

1. 消費者

MySQL テーブルに挿入されたデータを取得します。

ここでは、MySQL でデータを取得しています。

2. Ksql

Mysql テーブルに挿入され、Kafka トピックに反映されるインストリーム データを入力する必要があります。

ksql で期待される結果が得られません

このデータ パイプラインを手伝ってください。

ksql ストリームを使用したメッセージ処理があり、処理が終了した時刻を取得するために、処理された各行にタイムスタンプを追加したいと考えています。

最初は ROWTIME が更新されると思っていましたが、最初の Kafka トピックのメッセージから保持されているようです。

TIMESTAMP メソッドは、入力データを特定のタイムスタンプに変換するために使用されているようです。