kafka實現消息過濾

① 如何對Kafka進行監控

Jafka/Kafka Kafka是Apache下的一個子項目，是一個高性能跨語言分布式Publish/Subscribe消息隊列系統，而Jafka是在Kafka之上孵化而來的，即Kafka的一個升級版。具有以下特性：快速持久化，可以在O(1)的系統開銷下進行消息持久化；高吞吐，在一台

② spark讀取kafka數據，之後進行過濾，過濾後計算有多少條數據怎麼寫

spark讀取kafka數據，之後進行過濾，過濾後計算有多少條數據怎麼寫
前面應該還有個數據生產者，比如flume. flume負責生產數據，發送至kafka。 spark streaming作為消費者，實時的從kafka中獲取數據進行計算。 計算結果保存至redis，供實時推薦使用。 flume+kafka+spark+redis是實時數據收集與計算的一套經典架構

③ 怎麼對kafka中消費者拉取的數據進行過濾篩選

在消費端,對特定主題數據進行分組獲取,然後在獲取過程中對符合業務條件的數據進行處理,否則跳過,但還是會告訴kafka我已經消費過了。 示例代碼: 生產端: public...

④ kafka實現的是消息隊列的什麼協議

Kafka 分布式消息隊列 類似產品有JBoss、MQ 一、由Linkedln 開源，使用scala開發，有如下幾個特點： （1）高吞吐 （2）分布式 （3）支持多語言客戶端 （C++、Java） 二、組成: 客戶端是 procer 和 consumer，提供一些API，伺服器端是Broker，...

⑤ 如何實現kafka高並發發送消息

幾點需要注意： 盡量使用緩存，包括用戶緩存，信息緩存等，多花點內存來做緩存，可以大量減少與資料庫的交互，提高性能。 用jprofiler等工具找出性能瓶頸，減少額外的開銷。 優化資料庫查詢語句，減少直接使用hibernate等工具的直接生成語句（僅

⑥ 要怎麼處理才可以讓所有的節點並行處理kafka數據

首先，broker按照請求被發送的順序處理請求，並且按照同樣的順序發送響應。因為Kafka對消息的順序性有如下的保證：
Messages sent by a procer to a particular topic partition will be appended in the order they are sent. That is, if a message M1 is sent by the same procer as a message M2, and M1 is sent first, then M1 will have a lower offset than M2 and appear earlier in the log.

⑦ rabbitmq與kafka到底用哪個好

Kafka和RabbitMq一樣是通用意圖消息代理，他們都是以分布式部署為目的。但是他們對消息語義模型的定義的假設是非常不同的。我對"AMQP 更成熟"這個論點是持懷疑態度的。讓我們用事實說話來看看用什麼解決方案來解決你的問題。
a) 以下場景你比較適合使用Kafka。你有大量的事件(10萬以上/秒)、你需要以分區的，順序的，至少傳遞成功一次到混雜了在線和打包消費的消費者、你希望能重讀消息、你能接受目前是有限的節點級別高可用或則說你並不介意通過論壇/IRC工具得到還在幼兒階段的軟體的支持。
b) 以下場景你比較適合使用RabbitMQ。你有較少的事件（2萬以上/秒）並且需要通過復雜的路由邏輯去找到消費者、你希望消息傳遞是可靠的、你並不關心消息傳遞的順序、你需要現在就支持集群-節點級別的高可用或則說你需要7*24小時的付費支持（當然也可以通過論壇/IRC工具）。

⑧ Kafka有個什麼參數可以讓consumer阻塞知道新消息到達

）的方式從broker中獲取消息，但Pull有個缺

⑨ linux 怎樣查看kafka的某 topic數據

1、創建一個需要增加備份因子的topic列表的文件，文件格式是json格式的。

注意事項：

Kafka的目的是通過Hadoop的並行載入機制來統一線上和離線的消息處理，也是為了通過集群來提供實時的消息。

⑩ 如何保證kafka 的消息機制 ack-fail 源碼跟蹤

Kafka is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system, but with a unique design.(Kafka布式、區(partitioned)、基於備份(replicated)commit-log存儲服務.提供類似於messaging system特性,設計實現完全同)kafka種高吞吐量布式發布訂閱消息系統特性：
（1）、通O(1)磁碟數據結構提供消息持久化種結構於即使數TB消息存儲能夠保持間穩定性能
（2）、高吞吐量：即使非普通硬體kafka支持每秒數十萬消息
（3）、支持通kafka伺服器消費機集群區消息
（4）、支持Hadoop並行數據載入
、用Kafka面自帶腳本進行編譯
載Kafka源碼面自帶gradlew腳本我利用編譯Kafka源碼：
1 # wget
2 # tar -zxf kafka-0.8.1.1-src.tgz
3 # cd kafka-0.8.1.1-src
4 # ./gradlew releaseTarGz
運行面命令進行編譯現異信息：
01 :core:signArchives FAILED
02
03 FAILURE: Build failed with an exception.
04
05 * What went wrong:
06 Execution failed for task ':core:signArchives'.
07 > Cannot perform signing task ':core:signArchives' because it
08 has no configured signatory
09
10 * Try:
11 Run with --stacktrace option to get the stack trace. Run with
12 --info or --debug option to get more log output.
13
14 BUILD FAILED
bug（）用面命令進行編譯
1 ./gradlew releaseTarGzAll -x signArchives
候編譯功（編譯程現）編譯程我指定應Scala版本進行編譯：
1 ./gradlew -PscalaVersion=2.10.3 releaseTarGz -x signArchives
編譯完core/build/distributions/面kafka_2.10-0.8.1.1.tgz文件網載直接用
二、利用sbt進行編譯
我同用sbt編譯Kafka步驟：
01 # git clone
02 # cd kafka
03 # git checkout -b 0.8 remotes/origin/0.8
04 # ./sbt update
05 [info] [SUCCESSFUL ] org.eclipse.jdt#core;3.1.1!core.jar (2243ms)
06 [info] downloading ...
07 [info] [SUCCESSFUL ] ant#ant;1.6.5!ant.jar (1150ms)
08 [info] Done updating.
09 [info] Resolving org.apache.hadoop#hadoop-core;0.20.2 ...
10 [info] Done updating.
11 [info] Resolving com.yammer.metrics#metrics-annotation;2.2.0 ...
12 [info] Done updating.
13 [info] Resolving com.yammer.metrics#metrics-annotation;2.2.0 ...
14 [info] Done updating.
15 [success] Total time: 168 s, completed Jun 18, 2014 6:51:38 PM
16
17 # ./sbt package
18 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
19 Getting Scala 2.8.0 ...
20 :: retrieving :: org.scala-sbt#boot-scala
21 confs: [default]
22 3 artifacts copied, 0 already retrieved (14544kB/27ms)
23 [success] Total time: 1 s, completed Jun 18, 2014 6:52:37 PM
於Kafka 0.8及版本需要運行命令：
01 # ./sbt assembly-package-dependency
02 [info] Loading project definition from /export1/spark/kafka/project
03 [warn] Multiple resolvers having different access mechanism configured with
04 same name 'sbt-plugin-releases'. To avoid conflict, Remove plicate project
05 resolvers (`resolvers`) or rename publishing resolver (`publishTo`).
06 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
07 [warn] Credentials file /home/wyp/.m2/.credentials does not exist
08 [info] Including slf4j-api-1.7.2.jar
09 [info] Including metrics-annotation-2.2.0.jar
10 [info] Including scala-compiler.jar
11 [info] Including scala-library.jar
12 [info] Including slf4j-simple-1.6.4.jar
13 [info] Including metrics-core-2.2.0.jar
14 [info] Including snappy-java-1.0.4.1.jar
15 [info] Including zookeeper-3.3.4.jar
16 [info] Including log4j-1.2.15.jar
17 [info] Including zkclient-0.3.jar
18 [info] Including jopt-simple-3.2.jar
19 [warn] Merging 'META-INF/NOTICE' with strategy 'rename'
20 [warn] Merging 'org/xerial/snappy/native/README' with strategy 'rename'
21 [warn] Merging 'META-INF/maven/org.xerial.snappy/snappy-java/LICENSE'
22 with strategy 'rename'
23 [warn] Merging 'LICENSE.txt' with strategy 'rename'
24 [warn] Merging 'META-INF/LICENSE' with strategy 'rename'
25 [warn] Merging 'META-INF/MANIFEST.MF' with strategy 'discard'
26 [warn] Strategy 'discard' was applied to a file
27 [warn] Strategy 'rename' was applied to 5 files
28 [success] Total time: 3 s, completed Jun 18, 2014 6:53:41 PM
我sbt面指定scala版本：
01 <!--
02 User: 往記憶
03 Date: 14-6-18
04 Time: 20:20
05 bolg:
06 本文址：/archives/1044
07 往記憶博客專注於hadoop、hive、spark、shark、flume技術博客量干貨
08 往記憶博客微信公共帳號：iteblog_hadoop
09 -->
10 sbt "++2.10.3 update"
11 sbt "++2.10.3 package"
12 sbt "++2.10.3 assembly-package-dependency"