Beego の Kafka と Spark Streaming を使用したリアルタイム データ処理

インターネットおよびモノのインターネット技術の継続的な発展に伴い、私たちの生産や生活の中で生成されるデータの量は増加しています。このデータは、企業のビジネス戦略と意思決定において非常に重要な役割を果たします。このデータをより有効に活用するために、リアルタイムのデータ処理は企業や科学研究機関の日常業務の重要な部分となっています。この記事では、リアルタイム データ処理のために Beego フレームワークで Kafka と Spark Streaming を使用する方法を検討します。

1. Kafka とは

Kafka は、大量のデータを処理するために使用される高スループットの分散メッセージ キュー システムです。 Kafka は、メッセージ データを複数のトピックに分散して保存し、迅速に取得して配布できます。データ ストリーミング シナリオでは、Kafka は最も人気のあるオープン ソース メッセージング システムの 1 つとなり、LinkedIn、Netflix、Twitter などの多くのテクノロジー企業で広く使用されています。

2. Spark Streaming とは

Spark Streaming は、Apache Spark エコシステムのコンポーネントであり、データ ストリームのリアルタイム バッチ処理を実行できるストリーミング コンピューティング フレームワークを提供します。 Spark Streaming は拡張性と耐障害性が高く、複数のデータ ソースをサポートできます。 Spark Streaming を Kafka などのメッセージ キュー システムと組み合わせて使用​​すると、ストリーミング コンピューティング機能を実装できます。

3. Beego で Kafka と Spark Streaming を使用してリアルタイム データ処理を行う

Beego フレームワークを使用してリアルタイム データ処理を行う場合、Kafka と Spark Streaming を組み合わせてデータ受信を実現できます。そして加工。以下は、単純なリアルタイム データ処理プロセスです:

1. Kafka を使用してメッセージ キューを確立し、データをメッセージにカプセル化し、Kafka に送信します。
2. Spark Streaming を使用してストリーミング アプリケーションを構築し、Kafka メッセージ キュー内のデータをサブスクライブします。
3. サブスクライブされたデータについては、データクリーニング、データ集計、ビジネス計算など、さまざまな複雑な処理操作を実行できます。
4. 処理結果を Kafka に出力するか、ユーザーに視覚的に表示します。

以下、上記の処理を実装する方法を詳しく紹介します。

1. Kafka メッセージ キューを確立する

まず、Kafka パッケージを Beego に導入する必要があります。Go 言語で sarama パッケージを使用し、次のコマンドで取得できます:

go get gopkg.in/Shopify/sarama.v1

次に、Beego で Kafka メッセージ キューを確立し、生成されたデータを Kafka に送信します。サンプル コードは次のとおりです。

func initKafka() (err error) {

//配置Kafka连接属性
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Partitioner = sarama.NewRandomPartitioner
config.Producer.Return.Successes = true
//创建Kafka连接器
client, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
if err != nil {
    fmt.Println("failed to create producer, err:", err)
    return
}
//异步关闭Kafka
defer client.Close()
//模拟生成数据
for i := 1; i < 5000; i++ {
    id := uint32(i)
    userName := fmt.Sprintf("user:%d", i)
    //数据转为byte格式发送到Kafka
    message := fmt.Sprintf("%d,%s", id, userName)
    msg := &sarama.ProducerMessage{}
    msg.Topic = "test" //topic消息标记
    msg.Value = sarama.StringEncoder(message) //消息数据
    _, _, err := client.SendMessage(msg)
    if err != nil {
        fmt.Println("send message failed:", err)
    }
    time.Sleep(time.Second)
}
return

}

上記のコードでは、Sarama パッケージの SyncProducer メソッドを使用しています。 Kafka コネクタを作成し、必要な接続プロパティを設定します。次に、for ループを使用してデータを生成し、生成されたデータをメッセージにカプセル化して Kafka に送信します。

2. リアルタイム データ処理に Spark ストリーミングを使用する

リアルタイム データ処理に Spark ストリーミングを使用する場合、Spark と Kafka をインストールして構成する必要があります。次のコマンド:

sudo apt-get install spark

sudo apt-get installzookeeper

sudo apt-get install kafka

インストール完了後、 Spark Streaming を Beego パッケージに導入する必要があります:

import org.apache.spark.SparkConf

import org.apache.spark.streaming.{Seconds, StreamingContext}

import org.apache.spark.streaming.kafka.KafkaUtils

次に、データ ストリームを処理する必要があります。次のコードは、Kafka からデータを受信し、各メッセージを処理するロジックを実装します。

func main() {

//创建SparkConf对象
conf := SparkConf().setAppName("test").setMaster("local[2]")
//创建StreamingContext对象，设置1秒钟处理一次
ssc := StreamingContext(conf, Seconds(1))
//从Kafka中订阅test主题中的数据
zkQuorum := "localhost:2181"
group := "test-group"
topics := map[string]int{"test": 1}
directKafkaStream, err := KafkaUtils.CreateDirectStream(ssc, topics, zkQuorum, group)
if err != nil {
    panic(err)
}
lines := directKafkaStream.Map(func(message *sarama.ConsumerMessage) (string, int) {
    //从消息中解析出需要的数据
    data := message.Value
    arr := strings.Split(string(data), ",")
    id, _ := strconv.Atoi(arr[0])
    name := arr[1]
    return name, 1
})
//使用reduceByKey函数对数据进行聚合计算
counts := lines.ReduceByKey(func(a, b int) int {
    return a + b
})
counts.Print() 
//开启流式处理
ssc.Start()
ssc.AwaitTermination()

}

上記のコードでは、SparkConf を使用します。メソッドと StreamingContext メソッドを使用して、Spark Streaming コンテキストを作成し、データ ストリームの処理時間間隔を設定します。次に、Kafka メッセージ キュー内のデータをサブスクライブし、Map メソッドを使用して受信メッセージから必要なデータを解析し、次に ReduceByKey メソッドを使用してデータ集計計算を実行します。最後に、計算結果がコンソールに出力されます。

4. 概要

この記事では、Beego フレームワークで Kafka と Spark Streaming を使用してリアルタイム データ処理を行う方法を紹介します。 Kafka メッセージ キューを確立し、Spark Streaming を使用してデータ ストリームを処理することにより、合理化された効率的なリアルタイム データ処理プロセスを実現できます。この処理方法はさまざまな分野で広く利用されており、企業の意思決定に重要な参考となります。

以上がBeego の Kafka と Spark Streaming を使用したリアルタイム データ処理の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。