大數據技術與架構點擊右側關注，大數據開發領域最強公眾號！

暴走大數據點擊右側關注，暴走大數據！

Apache Kafka 是一個可擴展，高性能，低延遲的平臺，允許我們像消息系統一樣讀取和寫入數據。我們可以很容易地在 Java 中使用 Kafka。

Spark Streaming 是 Apache Spark 的一部分，是一個可擴展、高吞吐、容錯的實時流處理引擎。雖然是使用 Scala 開發的，但是支持 Java API。

Apache Cassandra 是分布式的 NoSQL 數據庫。在這篇文章中，我們將介紹如何通過這三個組件構建一個高擴展、容錯的實時數據處理平臺。

準備

在進行下面文章介紹之前，我們需要先創建好 Kafka 的主題以及 Cassandra 的相關表，具體如下：

在 Kafka 中創建名為 messages 的主題

$KAFKA_HOME$\bin\windows\kafka-topics.bat --create \
--zookeeper localhost:2181 \
--replication-factor 1 --partitions 1 \
--topic messages

在 Cassandra 中創建 KeySpace 和 table

CREATE KEYSPACE vocabulary
WITH REPLICATION = {
'class' : 'SimpleStrategy',
'replication_factor' : 1
};
USE vocabulary;
CREATE TABLE words (word text PRIMARY KEY, count int);

上面我們創建了名為 vocabulary 的 KeySpace，以及名為 words 的表。

添加依賴

我們使用 Maven 進行依賴管理，這個項目使用到的依賴如下：

org.apache.spark
spark-core_2.11
2.3.0
provided


org.apache.spark
spark-sql_2.11
2.3.0
provided


org.apache.spark
spark-streaming_2.11
2.3.0
provided


org.apache.spark
spark-streaming-kafka-0-10_2.11
2.3.0


com.datastax.spark
spark-cassandra-connector_2.11
2.3.0


com.datastax.spark
spark-cassandra-connector-java_2.11
1.5.2

數據管道開發

我們將使用 Spark 在 Java 中創建一個簡單的應用程序，它將與我們之前創建的Kafka主題集成。應用程序將讀取已發布的消息并計算每條消息中的單詞頻率。然后將結果更新到 Cassandra 表中。整個數據架構如下：

現在我們來詳細介紹代碼是如何實現的。

獲取 JavaStreamingContext

Spark Streaming 中的切入點是 JavaStreamingContext，所以我們首先需要獲取這個對象，如下：

SparkConf sparkConf = new SparkConf();
sparkConf.setAppName("WordCountingApp");
sparkConf.set("spark.cassandra.connection.host", "127.0.0.1");

JavaStreamingContext streamingContext = new JavaStreamingContext(
sparkConf, Durations.seconds(1));

從 Kafka 中讀取數據

有了 JavaStreamingContext 之后，我們就可以從 Kafka 對應主題中讀取實時流數據，如下：

Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "localhost:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", false);
Collection<String> topics = Arrays.asList("messages");

JavaInputDStreamString, String>> messages =
KafkaUtils.createDirectStream(
streamingContext,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String> Subscribe(topics, kafkaParams));

我們在程序中提供了 key 和 value 的 deserializer。這個是 Kafka 內置提供的。我們也可以根據自己的需求自定義 deserializer。

處理 DStream

我們在前面只是定義了從 Kafka 中哪張表中獲取數據，這里我們將介紹如何處理這些獲取的數據：

JavaPairDStream<String, String> results = messages
.mapToPair(
record -> new Tuple2<>(record.key(), record.value())
);
JavaDStream<String> lines = results
.map(
tuple2 -> tuple2._2()
);
JavaDStream<String> words = lines
.flatMap(
x -> Arrays.asList(x.split("\\s+")).iterator()
);
JavaPairDStream<String, Integer> wordCounts = words
.mapToPair(
s -> new Tuple2<>(s, 1)
).reduceByKey(
(i1, i2) -> i1 + i2
);

將數據發送到 Cassandra 中

最后我們需要將結果發送到 Cassandra 中，代碼也很簡單。

wordCounts.foreachRDD(
javaRdd -> {
Map<String, Integer> wordCountMap = javaRdd.collectAsMap();
for (String key : wordCountMap.keySet()) {
List wordList = Arrays.asList(new Word(key, wordCountMap.get(key)));
JavaRDD rdd = streamingContext.sparkContext().parallelize(wordList);
javaFunctions(rdd).writerBuilder("vocabulary", "words", mapToRow(Word.class)).saveToCassandra();
}
}
);

啟動應用程序

最后，我們需要將這個 Spark Streaming 程序啟動起來，如下：

streamingContext.start();
streamingContext.awaitTermination();

使用 Checkpoints

在實時流處理應用中，將每個批次的狀態保存下來通常很有用。比如在前面的例子中，我們只能計算單詞的當前頻率，如果我們想計算單詞的累計頻率怎么辦呢？這時候我們就可以使用 Checkpoints。新的數據架構如下

為了啟用 Checkpoints，我們需要進行一些改變，如下：

streamingContext.checkpoint("./.checkpoint");

這里我們將 checkpoint 的數據寫入到名為 .checkpoint 的本地目錄中。但是在現實項目中，最好使用 HDFS 目錄。

現在我們可以通過下面的代碼計算單詞的累計頻率：

JavaMapWithStateDStream> cumulativeWordCounts = wordCounts
.mapWithState(
StateSpec.function((word, one, state) -> {
int sum = one.orElse(0) + (state.exists() ? state.get() : 0);
Tuple2 output = new Tuple2<>(word, sum);
state.update(sum);return output;
}
)
);

部署應用程序

最后，我們可以使用 spark-submit 來部署我們的應用程序，具體如下：

$SPARK_HOME$\bin\spark-submit \
--class com.baeldung.data.pipeline.WordCountingAppWithCheckpoint \
--master local[2]
\target\spark-streaming-app-0.0.1-SNAPSHOT-jar-with-dependencies.jar

最后，我們可以在 Cassandra 中查看到對應的表中有數據生成了。完整的代碼可以參見 https://github.com/eugenp/tutorials/tree/master/apache-spark

歡迎點贊+收藏+轉發朋友圈素質三連

文章不錯？點個【在看】吧！??

總結

以上是生活随笔為你收集整理的maven依赖 spark sql_使用Kafka+Spark+Cassandra构建实时处理引擎的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。