apache-kafka

1熱

1答えて

Kafkaトピックでレコードが利用できない場合、StreamingQueryStatus.isDataAvailableがオンになるのはなぜですか？

私はspark構造化ストリーミングクエリを停止する必要があります。まずは、カフカで処理できるデータがあるかどうかチェックします。はいの場合は処理してからクエリを停止します。 if (query.status.isDataAvailable) { query.processAllAvailable() } query.stop() 私はJacek Lasko

0熱

1答えて

Logstashは私の構造は、このようなものですカフカ

から消費することはできません。まず、Filebeatはカフカへのメッセージを生成することができますし、私が使用してそれを確認することができます： kafka-console-consumer.bat --zookeeper localhost:2181 --topic {topicName} --from-beginning そしてそれはまた、コマンドによって消費することができます： kaf

0熱

1答えて

カフカブローカーと飼育係の間にどの程度の比率を保つ必要がありますか？

私は4カフカクラスターを持っているなら、なぜ5KKKを必要とするのか読んでいます。これは、カフカクラスターを作るのが最も楽観的な方法だと言われていますが、私はこの点を知りませんでした。私にこれを説明してください。私は今まで3つのブローカーと3つのZKを使ってみました。

-1熱

1答えて

3ノードカフカクラスタを設定するには

3つのノードカフカクラスタを設定し、ソースDBの数に30のトピックを接続する必要があります。処理するメッセージの正確な数はわかりませんが、1日あたり何百万もの数があると思います。カフカの保存期間は30日に設定されています。必要に応じて30日前のデータに戻すことができます。どうすれば進めてください。多くのおかげ

0熱

1答えて

Avroシリアライザとデシリアライザkafka java api

Kafka Avroシリアライザとデシリアライザは動作していません。私はkafkaコンソールのコンシューマーを使ってメッセージを消費しようとしました。メッセージが公開されているのがわかりました。 public class AvroProducer<T> { private static Properties props; static { props = new

3熱

2答えて

Spark Structured Streamingでの書き込み時のKafkaオフセットのキャプチャ

Spark 2.2のSpark Structuredストリーミングを使用して、HDFSディレクトリからKafkaトピックにストリームファイルを配信しています。私は、このトピックに書いているデータのカフカオフセットをキャプチャしたいと思います。私はカフカに書き込むには val write = jsonDF .writeStream.format("kafka") .option("checkp

0熱

1答えて

Camel：Kafkaの間違ったIPを検出する

私はRouteBuilderのソースとしてKafkaを設定しています。私の目標は、カフカの切断問題を処理することです。私のRouteBuilderは次のとおりです： new RouteBuilder() { public void configure() { onException(Exception.class).process(exchange -> {

0熱

1答えて

Kafka - 10行以上の読み方

データベースからjdbcを読み込んでSparkアプリケーションから消費するコネクタを作成しました。アプリケーションはデータベースのデータをよく読みますが、最初の10行しか読み込まれず、残りの部分は無視されるようです。どのようにして休息を取るべきなので、すべてのデータを計算できます。ここでは私の火花コードです： val brokers = "http://127.0.0.1:9092" val

2熱

1答えて

正規表現/パターンマッチングのルックアップテーブルにKafkaストリームを結合する

私の質問のタイトルをより明確にするために、まず私の質問の目標を述べます。私の目標は私の目標です：私の目標は、IntegerをIDとし、値としてStringを持つデータストリーム（KStream）を持つことです。この例では、値に誰かの名前を格納しています。次に、用語をキーとして格納するGlobalKTable（ストリームの値と一致する用語）とブール値フラグ（この用語に一致する名前はすべて「スパマー

0熱

1答えて

kafkaストリームのスパークアプリケーションが推奨を生成するのに時間がかかります

私はスパークアプリケーションのデータストリームをカフカストリームから読み込んでいます。私の要求は、ユーザーがリクエスト（検索/ブラウズなど）を行ったときにユーザーに製品の推奨事項を提示することです。私はすでにユーザーのスコアを含む訓練されたモデルを持っています。私はJavaとorg.apache.spark.mllib.recommendation.MatrixFactorizationMode