2023學年完整公開課版將Kafka與Hadoop和Storm集成

主講：朱佳目錄0102與Hadoop集成與Storm集成與Hadoop集成1與Hadoop集成可以使用Kafka構(gòu)建數(shù)據(jù)流水線，實現(xiàn)實時處理，并將數(shù)據(jù)加載到Hadoop進行離線處理。Hadoop生產(chǎn)者向Kafka集群發(fā)送HDFS數(shù)據(jù)，Hadoop消費者從Kafka集群將數(shù)據(jù)提取到HDFS。Kafka提供了創(chuàng)建Hadoop生產(chǎn)者和消費者的源代碼。還可以使用第三方解決方案來集成Kafka和Hadoop。與Hadoop集成下面來了解一下Kafka提供的Hadoop生產(chǎn)者和消費者的工作原理:■Hadoop生產(chǎn)者用于將數(shù)據(jù)從Hadoop集群發(fā)布到Kafka集群。將數(shù)據(jù)從Hadoop發(fā)送到Kafka的一種方法是，使用Avro二進制格式為Kafka生產(chǎn)者編寫Pig腳本，其中一行是一個消息，并將數(shù)據(jù)推送到Kafka集群。在這個方法中，可以使用AvroKafkaStorage生產(chǎn)者在同一個Pig腳本中寫入各種主題和協(xié)調(diào)器。另一種方法是使用KafkaOutputFormat類將HDFS數(shù)據(jù)發(fā)布到Kafka集群。Kafka的OutputFormat類繼承了Hadoop的OutputFormat類，所以在向Hadoop集群寫入消息時，它能夠自動將數(shù)據(jù)寫入Kafka.與Hadoop集成■Hadoop消費者將數(shù)據(jù)從Kafka移動到Hadoop。消費者是一個Hadoop任務(wù)，它從Kafka協(xié)調(diào)器獲取數(shù)據(jù)，并將數(shù)據(jù)并行加載到HDFS與Storm集成2與Storm集成ApacheStorm使用稱為spouts和bolts的實體來完成工作。spouts是日志數(shù)據(jù)連續(xù)流的來源，bolts消費spouts發(fā)送給它們的數(shù)據(jù)，Storm拓撲定義了數(shù)據(jù)如何從spouts流向bolts。與集成Hadoop集群一樣，需要同時運行兩個集群:Kafka集群和Storm集群。與Storm集成可以使用ApacheKafka提供的KafkaSpout代碼將Storm與Kafka集群集成。KafkaSpout連續(xù)從Kafka集群讀取數(shù)據(jù)。KafkaSpout在ZooKeeper的根路徑中存儲消息偏移量和其他消息消費跟蹤信息。在默認情況下，此消息偏移量存儲在Storm集群的ZooKeeperznodes中。還可以通過配置KafkaSpout將數(shù)據(jù)存儲在其他ZooKeeper集群中。與Storm集成需要提供KafkaSpout連接的配置信息來連接到Kafka集群。以下是必要的配置參數(shù):■Kafka協(xié)調(diào)器列表■每個主機的分區(qū)數(shù)■用于拉取消息的主題名稱■KafkaSpout用于存儲消費者偏移量的ZooKeeper根路徑■消費者的ID,用于區(qū)分ZooKeeper中存儲的消費者偏移量與Storm集成具備了所有配置參數(shù)后，可以按照以下方式初始化KafkaSpout類實例:spoutConfigspoutconfig=newSpoutConfig(ImmutableList.of("localhost:9092"，"localhost:9093"),2，:"testtopic","/kafk