Ứng dụng thời gian thực (Twitter)
Hãy để chúng tôi phân tích một ứng dụng theo thời gian thực để nhận các nguồn cấp dữ liệu twitter mới nhất và các thẻ bắt đầu bằng # của nó. Trước đó, chúng ta đã thấy sự tích hợp của Storm và Spark với Kafka. Trong cả hai tình huống, chúng tôi đã tạo Kafka Producer (sử dụng cli) để gửi thông điệp đến hệ sinh thái Kafka. Sau đó, tương tác giữa bão và tia lửa đọc các thông điệp bằng cách sử dụng người tiêu dùng Kafka và đưa nó vào hệ sinh thái bão và tia lửa tương ứng. Vì vậy, trên thực tế, chúng ta cần tạo một Kafka Producer, điều này sẽ -
- Đọc các nguồn cấp dữ liệu twitter bằng cách sử dụng “Twitter Streaming API”,
- Xử lý nguồn cấp dữ liệu,
- Trích xuất HashTags và
- Gửi nó cho Kafka.
Khi Kafka nhận được HashTags
, tích hợp Storm / Spark sẽ nhận thông tin và gửi nó đến hệ sinh thái Storm / Spark.
API phát trực tuyến trên Twitter
"Twitter Streaming API" có thể được truy cập bằng bất kỳ ngôn ngữ lập trình nào. “Twitter4j” là một thư viện Java không chính thức, mã nguồn mở, cung cấp một mô-đun dựa trên Java để dễ dàng truy cập vào “Twitter Streaming API”. “Twitter4j” cung cấp một khung dựa trên người nghe để truy cập các tweet. Để truy cập “API phát trực tuyến Twitter”, chúng tôi cần đăng nhập vào tài khoản nhà phát triển Twitter và sẽ nhận được các thông tin sauOAuth chi tiết xác thực.
- Customerkey
- CustomerSecret
- AccessToken
- AccessTookenSecret
Sau khi tài khoản nhà phát triển được tạo, hãy tải xuống tệp jar “twitter4j” và đặt nó vào đường dẫn lớp java.
Mã hóa nhà sản xuất Twitter Kafka hoàn chỉnh (KafkaTwitterProductioner.java) được liệt kê bên dưới:
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("Got a status deletion notice id:"
+ statusDeletionNotice.getStatusId());
}
@Override
public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
// System.out.println("Got track limitation notice:" +
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("Got stall warning:" + warning);
}
@Override
public void onException(Exception ex) {
ex.printStackTrace();
}
};
twitterStream.addListener(listener);
FilterQuery query = new FilterQuery().track(keyWords);
twitterStream.filter(query);
Thread.sleep(5000);
//Add Kafka producer config settings
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();
}
}Tổng hợp
Biên dịch ứng dụng bằng lệnh sau:
javac -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:. KafkaTwitterProducer.javaChấp hành
Mở hai bảng điều khiển. Chạy ứng dụng đã biên dịch ở trên như được hiển thị bên dưới trong một bảng điều khiển.
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 foodChạy bất kỳ ứng dụng nào trong số các ứng dụng Spark / Storm được giải thích trong chương trước trong một bản win-down khác. Điểm chính cần lưu ý là chủ đề được sử dụng phải giống nhau trong cả hai trường hợp. Ở đây, chúng tôi đã sử dụng “chủ đề đầu tiên của tôi” làm tên chủ đề.
Đầu ra
Đầu ra của ứng dụng này sẽ phụ thuộc vào các từ khóa và nguồn cấp dữ liệu hiện tại của twitter. Đầu ra mẫu được chỉ định bên dưới (tích hợp bão).
. . .
food : 1
foodie : 2
burger : 1
. . .