课程表

Apache Kafka课程

工具箱
速查手册

Kafka 实时应用程序

当前位置:免费教程 » 大数据/云 » Apache Kafka

让我们分析一个实时应用程序,以获取最新的Twitter Feed和其标签。 早些时候,我们已经看到了Storm和Spark与Kafka的集成。 在这两种情况下,我们创建了一个Kafka生产者(使用cli)向Kafka生态系统发送消息。 然后,stormspark集成通过使用Kafka消费者读取消息,并将其分别注入到storm和spark生态系统中。 因此,实际上我们需要创建一个Kafka Producer,

  • 使用“Twitter Streaming API"阅读Twitter Feed,
  • 处理Feeds,
  • 提取HashTags
  • 发送到Kafka。

一旦Kafka接收到 HashTags ,Storm / Spark集成接收到该信息并将其发送到Storm / Spark生态系统。

Twitter Streaming API

“Twitter Streaming API"可以使用任何编程语言访问。 “twitter4j"是一个开源的非官方Java库,它提供了一个基于Java的模块,可以轻松访问“Twitter Streaming API"。 “twitter4j"提供了一个基于监听器的框架来访问tweet。 要访问“Twitter Streaming API",我们需要登录Twitter开发者帐户,并应获取以下 OAuth 身份验证详细信息。

  • Customerkey
  • CustomerSecret
  • AccessToken
  • AccessTookenSecret

创建开发人员帐户后,下载“twitter4j"jar文件并将其放置在java类路径中。

完整的Twitter Kafka生产者编码(KafkaTwitterProducer.java)如下所列 -

  1. import java.util.Arrays;
  2. import java.util.Properties;
  3. import java.util.concurrent.LinkedBlockingQueue;
  4.  
  5. import twitter4j.*;
  6. import twitter4j.conf.*;
  7.  
  8. import org.apache.kafka.clients.producer.Producer;
  9. import org.apache.kafka.clients.producer.KafkaProducer;
  10. import org.apache.kafka.clients.producer.ProducerRecord;
  11.  
  12. public class KafkaTwitterProducer {
  13. public static void main(String[] args) throws Exception {
  14. LinkedBlockingQueue<Status> queue = new LinkedBlockingQueue<Sta-tus>(1000);
  15. if(args.length < 5){
  16. System.out.println(
  17. "Usage: KafkaTwitterProducer <twitter-consumer-key>
  18. <twitter-consumer-secret> <twitter-access-token>
  19. <twitter-access-token-secret>
  20. <topic-name> <twitter-search-keywords>");
  21. return;
  22. }
  23. String consumerKey = args[0].toString();
  24. String consumerSecret = args[1].toString();
  25. String accessToken = args[2].toString();
  26. String accessTokenSecret = args[3].toString();
  27. String topicName = args[4].toString();
  28. String[] arguments = args.clone();
  29. String[] keyWords = Arrays.copyOfRange(arguments, 5, arguments.length);
  30.  
  31. ConfigurationBuilder cb = new ConfigurationBuilder();
  32. cb.setDebugEnabled(true)
  33. .setOAuthConsumerKey(consumerKey)
  34. .setOAuthConsumerSecret(consumerSecret)
  35. .setOAuthAccessToken(accessToken)
  36. .setOAuthAccessTokenSecret(accessTokenSecret);
  37.  
  38. TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).get-Instance();
  39. StatusListener listener = new StatusListener() {
  40. @Override
  41. public void onStatus(Status status) {
  42. queue.offer(status);
  43.  
  44. // System.out.println("@" &plus; status.getUser().getScreenName()
  45. &plus; " - " &plus; status.getText());
  46. // System.out.println("@" &plus; status.getUser().getScreen-Name());
  47.  
  48. /*for(URLEntity urle : status.getURLEntities()) {
  49. System.out.println(urle.getDisplayURL());
  50. }*/
  51.  
  52. /*for(HashtagEntity hashtage : status.getHashtagEntities()) {
  53. System.out.println(hashtage.getText());
  54. }*/
  55. }
  56. @Override
  57. public void onDeletionNotice(StatusDeletionNotice statusDeletion-Notice) {
  58. // System.out.println("Got a status deletion notice id:"
  59. &plus; statusDeletionNotice.getStatusId());
  60. }
  61. @Override
  62. public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
  63. // System.out.println("Got track limitation notice:" &plus;
  64. num-berOfLimitedStatuses);
  65. }
  66.  
  67. @Override
  68. public void onScrubGeo(long userId, long upToStatusId) {
  69. // System.out.println("Got scrub_geo event userId:" &plus; userId &plus;
  70. "upToStatusId:" &plus; upToStatusId);
  71. }
  72. @Override
  73. public void onStallWarning(StallWarning warning) {
  74. // System.out.println("Got stall warning:" &plus; warning);
  75. }
  76. @Override
  77. public void onException(Exception ex) {
  78. ex.printStackTrace();
  79. }
  80. };
  81. twitterStream.addListener(listener);
  82. FilterQuery query = new FilterQuery().track(keyWords);
  83. twitterStream.filter(query);
  84.  
  85. Thread.sleep(5000);
  86. //Add Kafka producer config settings
  87. Properties props = new Properties();
  88. props.put("bootstrap.servers", "localhost:9092");
  89. props.put("acks", "all");
  90. props.put("retries", 0);
  91. props.put("batch.size", 16384);
  92. props.put("linger.ms", 1);
  93. props.put("buffer.memory", 33554432);
  94. props.put("key.serializer",
  95. "org.apache.kafka.common.serializa-tion.StringSerializer");
  96. props.put("value.serializer",
  97. "org.apache.kafka.common.serializa-tion.StringSerializer");
  98. Producer<String, String> producer = new KafkaProducer<String, String>(props);
  99. int i = 0;
  100. int j = 0;
  101. while(i < 10) {
  102. Status ret = queue.poll();
  103. if (ret == null) {
  104. Thread.sleep(100);
  105. i++;
  106. }else {
  107. for(HashtagEntity hashtage : ret.getHashtagEntities()) {
  108. System.out.println("Hashtag: " &plus; hashtage.getText());
  109. producer.send(new ProducerRecord<String, String>(
  110. top-icName, Integer.toString(j++), hashtage.getText()));
  111. }
  112. }
  113. }
  114. producer.close();
  115. Thread.sleep(5000);
  116. twitterStream.shutdown();
  117. }
  118. }

汇编

使用以下命令编译应用程序 -

  1. javac -cp "/path/to/kafka/libs/*":"/path/to/twitter4j/lib/*":. KafkaTwitterProducer.java

执行

打开两个控制台。 在一个控制台中运行上面编译的应用程序,如下所示。

  1. java -cp “/path/to/kafka/libs/*":"/path/to/twitter4j/lib/*":
  2. . KafkaTwitterProducer <twitter-consumer-key>
  3. <twitter-consumer-secret>
  4. <twitter-access-token>
  5. <twitter-ac-cess-token-secret>
  6. my-first-topic food

在另一个窗口中运行前一章中解释的Spark / Storm应用程序中的任何一个。 主要要注意的是,在这两种情况下使用的主题应该是相同的。 在这里,我们使用“我的第一主题"作为主题名称。

输出

此应用程序的输出将取决于关键字和Twitter的当前Feed。 下面指定样本输出(集成storm)。

  1. . . .
  2. food : 1
  3. foodie : 2
  4. burger : 1
  5. . . .
转载本站内容时,请务必注明来自W3xue,违者必究。
 友情链接:直通硅谷  点职佳  北美留学生论坛

本站QQ群:前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试,对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接(目前9元/月)请联系QQ:27243702 沸活量
皖ICP备17017327号-2 皖公网安备34020702000426号