apache-kafka

0熱度

2回答

最終目標：連接Elasticsearch和kafka，並將ES索引中正在進行的更改事件接收到kafka。從卡夫卡，我有聽衆做進一步處理。方法：我使用Logstash輸入和輸出插件。這是配置代碼。 input { elasticsearch { hosts => ["localhost:9200"] index => "liferay-20116"

0熱度

1回答

錯誤登記的Avro架構：「串」

代碼發送Avro的消息到卡夫卡的話題 props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONF

1熱度

1回答

卡夫卡生產者配額

這裏是我們的物聯網平臺的入站郵件傳遞流程： Device ---(MQTT)---> RabbitMQ Broker ---(AMQP)---> Apache Storm ---> Kafka 我期待實現解決方案，有效地限制/節流數據以每個發佈到卡夫卡每秒量客戶基礎。當前的策略利用Guava的RateLimiter，每個設備都有自己的本地緩存實例。當接收到設備消息時，映射到該deviceId

0熱度

1回答

Kafka Connect - 無法添加缺少的字段SinkRecordField {schema = Schema {BYTES}，name ='CreateUID'，isPrimaryKey = true}，

我正在使用JDBC源連接器從Teradata表中讀取數據並推送到Kafka主題。但是當我試圖使用JDBC宿連接器讀Kafka主題並推送到Oracle表時，它會引發下面的錯誤。我相信錯誤是因爲參數pk.mode和pk.fields，我不確定要使用什麼。我的terradata有一個主鍵UserID + DatabaseID。我已經在Oracle中創建了這個表，主鍵是Userid + databas

-3熱度

1回答

使用Scala/Spark編程計算卡夫卡主題中的消息數

需求是使用Scala或Spark編程計算卡夫卡主題中的消息數。我對這兩種編程都很陌生，所以我不確定該怎麼做。任何人都可以幫我編寫代碼或指導我如何實現。

1熱度

1回答

卡夫卡生產商很慢

我是卡夫卡新手，我有一個問題，我無法解決。我在我自己的計算機中安裝了Kafka和Zookeeper（不在Linux中），並且我創建了一個包含多個分區（在6和12分區之間播放）的主題的代理。當我創建消費者時，他們完美地工作並且以良好的速度閱讀，但是提到生產者，我創建了許多網站中可以看到的簡單生產者。生產者在一個循環內併發送很多短消息（大約2000個非常短的消息）。我可以看到消費者非常quicl

1熱度

1回答

如何使用增長文件作爲Apache Kafka製作者並只讀取新添加的數據

我正在嘗試將文件用作我的製作人。源文件連續增長（例如每秒記錄20條記錄）。下面是類似我的問題後： How to write a file to Kafka Producer 但是，在這種情況下，整個文件被讀取並添加到卡夫卡的話題每一個新行插入文件的時間。我只需要將新添加的行發送到主題（即，如果該文件已包含10行，並且附加了4行，則只需將這4行發送到該主題）。有沒有辦法實現這個？其它的解決方案的

0熱度

2回答

卡夫卡描述主題命令列表多個主題描述

對主題my.topic.test運行以下命令給出對my.topic.test和my.topic_test（注意名稱中的點和下劃線的區別）主題的描述。是期望的行爲還是錯誤（我會傾向於一個錯誤）？ kafka-topics --describe --zookeeper my_zookeeper_ip:2181 --topic my.topic.test Topic:my.topic.test P

0熱度

1回答

如何在卡夫卡流應用中的固定時間窗口中統計唯一用戶？

對於用戶在我們的平臺上執行的每個事件，我們都有一個唯一主題的kafka消息。每個事件/ kafka消息都有一個公共字段userId。我們現在想從這個話題知道我們每小時有多少獨特的用戶。所以我們對用戶的事件類型和個人數量不感興趣。我們只想知道每個小時有多少獨特用戶在使用。實現此目的的最簡單方法是什麼？我目前的想法似乎不是很簡單，看到這裏的僞代碼： stream .selectKey() // u

1熱度

1回答

來自Kafka主題的Spark Streaming拋出偏移超出範圍，無法重新啓動流

我在Spark 2.1.1上運行流式作業，輪詢Kafka 0.10。我正在使用Spark KafkaUtils類創建一個DStream，並且所有內容都正常工作，直到由於保留策略導致數據超出主題。如果任何數據超出了主題，我會停止工作做出一些更改，但我得到的錯誤表明我的偏移量超出範圍。我做了很多研究，包括查看火花源代碼，並且我看到很多評論，如本期的評論：SPARK-19680 - 基本上說數據不應該丟