Webアクセス解析や、ログのリアルタイムモニタリングと不正検知、ソーシャルメディア分析などの時に、オープンソースの分散ストリーミングプラットフォームと呼ばれるApache KafkaとSparkにストリームデータを処理するSpark Streamingを組み合わせたストリームデータ処理システムはよく利用されると考えられます。今回の記事はTwitterのメッセージ分析を例として、Alibaba CloudのE-MapReduce上で、KafkaとSpark Streamingの統合方法について、皆さんにご紹介させて頂きたいと思います。
検証環境について
Spark Streaming
- EMR-3.20.0
- クラスタータイプは Hadoop
- ハードウェア構成(Header)はecs.sn2.largeを1台
- ハードウェア構成(Worker)はecs.sn2.largeを2台
# cat /etc/redhat-release
CentOS Linux release 7.4.1708 (Core)
# uname -r
3.10.0-693.2.2.el7.x86_64
# hadoop version
Hadoop 2.8.5
Subversion Unknown -r f6ae72b25f6dbb9925f0850ac865274d31a28e30
Compiled by root on 2019-04-19T06:38Z
Compiled with protoc 2.5.0
From source with checksum 9624fc19bc23f1bbeacb1ae4bee88e7
This command was run using /opt/apps/ecm/service/hadoop/2.8.5-1.3.0/package/hadoop-2.8.5-1.3.0/share/hadoop/common/hadoop-common-2.8.5.jar
# spark-submit --version
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.4.2
/_/
Using Scala version 2.11.12, OpenJDK 64-Bit Server VM, 1.8.0_151
Branch branch-2.4.2
Compiled by user root on 2019-04-26T06:32:50Z
Revision ef3e6ff4b20dc86ac81291664bb0ed3cf2641fea
Url /root/git-repos/emr-spark.git/
Type --help for more information.
Kafka
- EMR-3.20.0
- Zookeeper 3.4.13
- Kakfa 1.1.1
- クラスタータイプは Kafka
- ハードウェア構成(Header)はecs.sn2.largeを1台
- ハードウェア構成(Worker)はecs.sn2.largeを2台
[root@emr-header-1 bin]# echo envi | nc localhost 2181
Environment:
zookeeper.version=3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 04:05 GMT
host.name=emr-header-1.cluster-43709
java.version=1.8.0_151
java.vendor=Oracle Corporation
java.home=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.151-1.b12.el7_4.x86_64/jre
java.class.path=/usr/lib/zookeeper-current/bin/../build/classes:/usr/lib/zookeeper-current/bin/../build/lib/*.jar:/usr/lib/zookeeper-current/bin/../lib/slf4j-log4j12-1.7.25.jar:/usr/lib/zookeeper-current/bin/../lib/slf4j-api-1.7.25.jar:/usr/lib/zookeeper-current/bin/../lib/netty-3.10.6.Final.jar:/usr/lib/zookeeper-current/bin/../lib/log4j-1.2.17.jar:/usr/lib/zookeeper-current/bin/../lib/jline-0.9.94.jar:/usr/lib/zookeeper-current/bin/../lib/audience-annotations-0.5.0.jar:/usr/lib/zookeeper-current/bin/../zookeeper-3.4.13.jar:/usr/lib/zookeeper-current/bin/../src/java/lib/*.jar:/etc/ecm/zookeeper-conf::/var/lib/ecm-agent/data/jmxetric-1.0.8.jar
java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib
java.io.tmpdir=/tmp
java.compiler=<NA>
os.name=Linux
os.arch=amd64
os.version=3.10.0-693.2.2.el7.x86_64
user.name=hadoop
user.home=/home/hadoop
user.dir=/home/hadoop
Hosebird Client
Hosebird Clientというのは、Kafka Producerと連携する時に、TwitterのStreaming APIをコールするJava Http Clientです。
詳しく知りたい方は下記のgithubにご参考頂ければと思います。
[https://github.com/twitter/hbc:embed:cite]
全体構成図
指定したキーワードがTwitterでどれだけ話題になったかをリアルタイムのツイート数で表現させたいという分析の目的から、
まずはじめに、構築するAlibaba Cloudの構成図は以下となります。
Kafka Producer
準備が整ったので、ローカルのJava開発環境で早速コードを書いていきましょう！
まずは、以下のサンプルコードを参考に、Kafka Producerを作成し、jarファイルを生成します。
Tweetsキーワード
"bitcoin"、"Blockchain"、"IoT"、"5G"、４つのキーワードを指定しました。
Kakfa BootstrapServers
Kafkaクラスターの任意の一台のIPアドレスで構いません。
Twitter Streaming API認証情報
Twitter Streaming APIを利用する為に、consumerKey、consumerSecret、token、secretをそれぞれ事前に取得して入力します。
public class ProducerTest {
Logger logger = LoggerFactory.getLogger(ProducerTest.class.getName());
/** ---------------------- Twitter Streaming API情報 ---------------------- */
String consumerKey = "xxxxxxxxxxxxxxxx";
String consumerSecret = "xxxxxxxxxxxxxxxx";
String token = "xxxxxxxxxxxxxxxx";
String secret = "xxxxxxxxxxxxxxxx";
String mytopic = "tweets_poc";
/** ---------------------- Tweetsキーワードを指定 ---------------------- */
List<String> terms = Lists.newArrayList("bitcoin","Blockchain","IoT","5G");
public ProducerTest(){}
public static void main(String[] args) {
new ProducerTest().run();
}
public void run(){
logger.info("Setup");
BlockingQueue<String> msgQueue = new LinkedBlockingQueue<String>(1000);
Client client = createTwitterClient(msgQueue);
client.connect();
KafkaProducer<String, String> producer = createKafkaProducer();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
logger.info("stopping application...");
logger.info("shutting down client from twitter...");
client.stop();
logger.info("closing producer...");
producer.close();
logger.info("done!");
}));
while (!client.isDone()) {
String msg = null;
try {
msg = msgQueue.poll(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
client.stop();
}
if (msg != null){
logger.info(msg);
if(StringUtils.containsIgnoreCase(msg,"Bitcoin")){
producer.send(new ProducerRecord<>(mytopic, "Bitcoin", msg), new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
logger.error("Something bad happened", e);
}
}
});
}
else if (StringUtils.containsIgnoreCase(msg,"Blockchain")) {
producer.send(new ProducerRecord<>(mytopic, "Blockchain", msg), new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
logger.error("Something bad happened", e);
}
}
});
}
else if (StringUtils.containsIgnoreCase(msg,"IoT")) {
producer.send(new ProducerRecord<>(mytopic, "IoT", msg), new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
logger.error("Something bad happened", e);
}
}
});
}
else if (StringUtils.containsIgnoreCase(msg,"5G")) {
producer.send(new ProducerRecord<>(mytopic,5,"5G", msg));
}
else{
producer.send(new ProducerRecord<>(mytopic, null, msg), new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
logger.error("Something bad happened", e);
}
}
});
}
}
}
logger.info("End of application");
}
/** ---------------------- Hosebird Clientを作成 ---------------------- */
public Client createTwitterClient(BlockingQueue<String> msgQueue){
Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
StatusesFilterEndpoint hosebirdEndpoint = new StatusesFilterEndpoint();
hosebirdEndpoint.trackTerms(terms);
Authentication hosebirdAuth = new OAuth1(consumerKey, consumerSecret, token, secret);
ClientBuilder builder = new ClientBuilder()
.name("Hosebird-Client-01")
.hosts(hosebirdHosts)
.authentication(hosebirdAuth)
.endpoint(hosebirdEndpoint)
.processor(new StringDelimitedProcessor(msgQueue));
Client hosebirdClient = builder.build();
return hosebirdClient;
}
/** ---------------------- kakfa producerを作成 ---------------------- */
public KafkaProducer<String, String> createKafkaProducer(){
String bootstrapServers = "xxxxxxxxxxxxxxxx";
Properties properties = new Properties();
properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
properties.setProperty(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE));
properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "20");
properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, Integer.toString(32*1024));
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
return producer;
}
}
Spark Streaming Consumer
Spark Streamingは、数秒から数分ほどの短い間隔で繰り返しバッチ処理というマイクロバッチ方式によるストリームデータ処理機能を提供します。E-MapreduceのKafka Brokerのバージョンは0.10.0以上である為、kafkaレシーバーを使わず、各Partitionから直接メッセージを読み込むというダイレクトのアプローチでKafkaと連携できています。今回は使った間隔データは下記のようになります。
DStreamのバッチ間隔 １秒
スライディング間隔 １秒
ウィンドウサイズ 300秒
ローカルのJava開発環境で、以下のサンプルコードを参考に、Spark Streaming Consumerを作成し、jarファイルを生成します。
public class SparkStreamingPoC {
private static JsonParser jsonParser = new JsonParser();
public static void main(String[] args) throws InterruptedException {
SparkConf sparkConf = new SparkConf().setAppName("TweetsApp");
JavaStreamingContext streamingContext = new JavaStreamingContext(
sparkConf, Durations.seconds(1));
/** ---------------------- kafkaパラメータ設定 ---------------------- */
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "xxxxxxxxxxxxxxxx");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "tweets_group");
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", false);
Collection<String> topics = Arrays.asList("tweets_poc");
/** ---------------------- Spark Stream作成 ---------------------- */
JavaInputDStream<ConsumerRecord<String, String>> stream =
KafkaUtils.createDirectStream(
streamingContext,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String> Subscribe(topics, kafkaParams));
JavaPairDStream<String,Integer> s1 = stream.mapToPair(record -> new Tuple2<>(record.key(),1));
JavaPairDStream<String,Integer> s2 = s1.reduceByKeyAndWindow((a,b) -> a + b,Durations.seconds(300),Durations.seconds(1));
JavaPairDStream<Integer,String> s3 = s2.mapToPair(Tuple2::swap);
JavaPairDStream<Integer,String> s4 = s3.transformToPair(rdd -> rdd.sortByKey(false));
s4.print();
streamingContext.start();
streamingContext.awaitTermination();
}
}
Sparkアプリを実行する
まず、作成したKafka ProducerとSpark Streaming ConsumerのJarファイルを一旦Alibaba Cloud OSSにアップロードしておきます。そして、ECSにsshログインして、ossutilなどのツールでKafka Producerのjarファイル(TweetsProducerTest-1.0-jar-with-dependencies.jar)をECSにダウンロードしていきます。ダウンロードできたら、下記のコマンドでKafka Producerを起動して、Twitterに投稿したメッセージを収集させます。
java -jar TweetsProducerTest-1.0-jar-with-dependencies.jar
E-MapReduceコンソールから、Kafkaに書き込んでいるTopicのPartitionやOffset、ISRなどの情報もリアルタイムで確認することもできます。
|
|---|
|
master引数
E-MapReduceはYarnモードを使用しますので、下図赤枠のCLIのアーギュメント(master)をYarnに指定する。
OSSプロトコル
ossref：OSSからjarファイルをダウンロードして実行する。
下図のように、ETL処理の結果も確認することができました。
最後
いかかでしたでしょうか
普通ならば、sshなどでmaster nodeや踏み台サーバーにログインし、コマンドラインからJobを実行するとの手順ですが、今回のように、E-MapReduceコンソールのデータプラットフォームより、Jobの実行はGUIベースで操作できるようになりました。さらに、Jobの結果はすぐに反映できる点はJobのトラブルシューティングにも役に立てると思います。
Spark Streamingとkafkaを組み合わせたストリームデータ処理システムをE-Mapreduce上に構築し、その検証結果を紹介しました。実は、汎用のリアルタイムデータ処理プラットフォームとして、今回ご紹介したApache Spark以外、Apache Flinkを利用することもできます。Flinkに慣れ親しんだ開発者は、Alibaba CloudのマネージドサービスRealtime Computeをご参考頂ければと思います。
また、Twitterメッセージ以外にも、日々多くの各種トラフィックログ、メールのデータなどをリアルタイムな分析やモニタリングを実現したい場合にも活用できるかと思いますので、是非使ってみてください！