kafka-producer-api

0熱度

1回答

我有下面的代碼，它將監聽卡夫卡主題並使用火花流傳輸原樣重現文本。但是，我無法在控制檯上看到文本。我沒有收到控制檯上的任何錯誤消息。我可能是錯的，但我期望卡夫卡主題中的文本顯示在控制檯上。 object scalaSparkProcessor { def main(args: Array[String]) { if (args.length < 3) { System.err.

0熱度

1回答

使用卡夫卡到兩個客戶端

我不得不機（機＃1和機＃2）之間卡夫卡集羣設置和配置之間傳輸文件如下： 1）每個機被配置成具有一個代理和一個動物園管理員正在運行。 2）服務器和動物園管理員屬性被配置爲具有多代理，多節點動物園管理員。我現在有KafkaProducer和KafkaConsumer以下認識： 1）如果我從machine#1發送文件到machine#2，它的分解在使用一些默認的分隔符（LF或\ n）的線。 2）因此，

4熱度

1回答

Kafka不會消耗所有生成的數據

我有一個安裝在具有8個內核和32GB RAM的VM上的單個kafka實例。我從10臺不同的機器寫入（生產），並從一臺機器中消耗，所有機器都在同一個網絡中。我生成的數據大小約爲35MBit/s。由於某些原因，大部分時間我不能消耗超過〜10MBit/s（在有限的時間段內，我設法消耗所有生成的數據），儘管kafka和消費者服務器大多數閒置（因此我不認爲這是保留問題）。 kafka可能會忽略一些生成

0熱度

2回答

Flink流 - 延遲和吞吐量檢測

我正在嘗試運行Flink流式作業。我想確定流式傳輸過程的吞吐量和延遲。我已經開始了卡夫卡經紀人服務器，並從卡夫卡傳入消息。我如何計算每秒消息（吞吐量）？（像rdd.count。有沒有類似的方法來獲取傳入消息的計數）（完整的scenerio：我已經通過生產者發送消息作爲Json對象，我添加了一些信息，如名稱作爲字符串和在Json對象中也是System.currentTimeMills 在流式傳輸

0熱度

1回答

Kafka Producer停止我的代碼

我正在調用一個從kafka生產者發送一些數據的函數，但是在它發送之後，我返回了一個不返回的響應。代碼在返回時卡住了。任何人有任何想法發生什麼事？我的代碼如下， def postEvent(eventData): print("The eventData is...",eventData) timestamp = datetime.now().__format__("%Y-%m

1熱度

2回答

卡夫卡流：一個記錄到多個記錄

鑑於：我在卡夫卡有兩個主題讓我們說主題A和主題B.卡夫卡流從主題A讀取記錄，處理它併產生多個記錄（比如說recordA和recordB）對應於消耗的記錄。現在，問題是如何使用Kafka Streams來實現這一點。 KStream<String, List<Message>> producerStreams[] = recordStream.mapValues(new ValueMapper<Me

0熱度

1回答

卡夫卡+星火流：1秒

EDIT2的恆定延遲：最後，我已經用Java做了我自己製作，而且運作良好，因此的問題是在卡夫卡的控制檯製片。卡夫卡控制檯消費者運作良好。編輯：我已經嘗試過版本0.9.0.1，並具有相同的行爲。我正在研究我的單身漢最終項目，Spark Streaming和Flink之間的比較。在兩個框架之前，我使用Kafka和一個腳本來生成數據（如下所述）。我的第一個測試是比較兩種框架與簡單工作負載之間的延遲，

0熱度

1回答

Kafka遠程製作人 - advertised.listeners

我在CDH 5.9上運行卡夫卡0.10.0，羣集被kerborized。我想要做的是將消息從遠程機器寫入我的卡夫卡經紀人。羣集（其中安裝了Kafka）具有內部以及外部IP地址。集羣中機器的主機名解析爲私有IP，遠程機器將相同的主機名解析爲公有IP地址。我從遠程機器打開必要的端口9092（我使用SASL_PLAINTEXT協議）到Kafka Broker，驗證了使用telnet。第一步 -

1熱度

2回答

org.apache.kafka.common.config.ConfigException：缺少必需的配置「bootstrap.servers」其中有

在運行我的製片人上課的時候Eclipse的我得到這個錯誤沒有默認值： org.apache .kafka.common.config.ConfigException：缺少必需的配置「bootstrap.servers」，它沒有默認值這裏是我的製作等級： public class SimpleProducer { public static void main(String[] arg

0熱度

1回答

類型不匹配：無法從Map <String，ConsumerRecords <String，Supplier >>轉換爲ConsumerRecords <String，供應商>

當我試圖在eclipse中運行我的SupplierConsumer類時，出現這些錯誤。這裏是我的代碼： public class SupplierConsumer{ public static void main(String[] args) throws Exception{ String topicName = "SupplierTopic"; Str