Apache Kafka 实时应用(推特)
让我们分析一个实时应用程序以获取最新的 Twitter 提要及其标签。早些时候，我们已经看到了 Storm 和 Spark 与 Kafka 的集成。在这两种情况下，我们都创建了一个 Kafka Producer(使用 cli)来向 Kafka 生态系统发送消息。然后，storm 和 spark 集成使用 Kafka 消费者读取消息，并将其分别注入到 Storm 和 Spark 生态系统中。所以，实际上我们需要创建一个 Kafka Producer，它应该：
- 使用“Twitter Streaming API”阅读 Twitter 提要，
- 处理提要，
- 提取 HashTags 和
- 发给Kafka 。
Once the
HashTags
由 Kafka 接收，Storm / Spark 集成接收信息并将其发送到 Storm / Spark 生态系统。
Twitter 流媒体 API
“Twitter Streaming API”可以用任何编程语言访问。 “twitter4j”是一个开源的、非官方的 Java 库，它提供了一个基于 Java 的模块来轻松访问“Twitter Streaming API”。 “twitter4j”提供了一个基于监听器的框架来访问推文。要访问“Twitter Streaming API”，我们需要登录 Twitter 开发者帐户，并且应该得到以下信息 OAuth 身份验证详细信息。
- 客户密钥
- 客户秘密
- 访问令牌
- AccessTookenSecret
创建开发者帐户后，下载“twitter4j”jar 文件并将其放在 java 类路径中。
完整的 Twitter Kafka 生产者编码(KafkaTwitterProducer.java)如下：
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;
import twitter4j.*;
import twitter4j.conf.*;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class KafkaTwitterProducer {
public static void main(String[] args) throws Exception {
LinkedBlockingQueue<Status> queue = new LinkedBlockingQueue<Sta-tus>(1000);
if(args.length < 5){
System.out.println(
"Usage: KafkaTwitterProducer <twitter-consumer-key>
<twitter-consumer-secret> <twitter-access-token>
<twitter-access-token-secret>
<topic-name> <twitter-search-keywords>");
return;
}
String consumerKey = args[0].toString();
String consumerSecret = args[1].toString();
String accessToken = args[2].toString();
String accessTokenSecret = args[3].toString();
String topicName = args[4].toString();
String[] arguments = args.clone();
String[] keyWords = Arrays.copyOfRange(arguments, 5, arguments.length);
ConfigurationBuilder cb = new ConfigurationBuilder();
cb.setDebugEnabled(true)
.setOAuthConsumerKey(consumerKey)
.setOAuthConsumerSecret(consumerSecret)
.setOAuthAccessToken(accessToken)
.setOAuthAccessTokenSecret(accessTokenSecret);
TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).get-Instance();
StatusListener listener = new StatusListener() {
@Override
public void onStatus(Status status) {
queue.offer(status);
// System.out.println("@" + status.getUser().getScreenName()
+ " - " + status.getText());
// System.out.println("@" + status.getUser().getScreen-Name());
/*for(URLEntity urle : status.getURLEntities()) {
System.out.println(urle.getDisplayURL());
}*/
/*for(HashtagEntity hashtage : status.getHashtagEntities()) {
System.out.println(hashtage.getText());
}*/
}
@Override
public void onDeletionNotice(StatusDeletionNotice statusDeletion-Notice) {
// System.out.println("得到一个状态删除通知id:"
+ statusDeletionNotice.getStatusId());
}
@Override
public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
// System.out.println("得到曲目限制通知：" +
num-berOfLimitedStatuses);
}
@Override
public void onScrubGeo(long userId, long upToStatusId) {
// System.out.println("Got scrub_geo event userId:" + userId +
"upToStatusId:" + upToStatusId);
}
@Override
public void onStallWarning(StallWarning warning) {
// System.out.println("得到停顿警告：" + warning);
}
@Override
public void onException(Exception ex) {
ex.printStackTrace();
}
};
twitterStream.addListener(listener);
FilterQuery query = new FilterQuery().track(keyWords);
twitterStream.filter(query);
Thread.sleep(5000);
// 添加 Kafka 生产者配置设置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer",
"org.apache.kafka.common.serializa-tion.StringSerializer");
props.put("value.serializer",
"org.apache.kafka.common.serializa-tion.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);
int i = 0;
int j = 0;
while(i < 10) {
Status ret = queue.poll();
if (ret == null) {
Thread.sleep(100);
i++;
}else {
for(HashtagEntity hashtage : ret.getHashtagEntities()) {
System.out.println("Hashtag: " + hashtage.getText());
producer.send(new ProducerRecord<String, String>(
top-icName, Integer.toString(j++), hashtage.getText()));
}
}
}
producer.close();
Thread.sleep(5000);
twitterStream.shutdown();
}
}
汇编
使用以下命令编译应用程序：
javac -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:. KafkaTwitterProducer.java
执行
打开两个控制台。在一个控制台中运行上面编译的应用程序，如下所示。
java -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”: . KafkaTwitterProducer <twitter-consumer-key> <twitter-consumer-secret> <twitter-access-token> <twitter-ac-cess-token-secret> my-first-topic food
在另一个窗口中运行上一章中解释的任何一个 Spark/Storm 应用程序。需要注意的要点是在两种情况下使用的主题应该相同。在这里，我们使用“my-first-topic”作为主题名称。
此应用程序的输出将取决于关键字和 twitter 的当前提要。下面指定了一个示例输出(风暴集成)。
. . . food : 1 foodie : 2 burger : 1 . . .