Apache Kafka と Spark Streaming によるストリーミング データ処理

こんにちは、ハブル！ 今日は、Spark Streaming を使用して Apache Kafka のメッセージ ストリームを処理し、処理結果を AWS RDS クラウド データベースに書き込むシステムを構築します。

ある信用機関が、そのすべての支店にわたって受信トランザクションを「その場で」処理するタスクを私たちに課したと想像してみましょう。 これは、財務省のオープン通貨ポジション、取引の限度額または財務結果などを迅速に計算する目的で行うことができます。

魔法や魔法の呪文を使用せずにこのケースを実装する方法 - カットの下をお読みください。 行く！

(画像出典)

導入

もちろん、大量のデータをリアルタイムで処理することは、最新のシステムで使用される機会を十分に提供します。 このための最も一般的な組み合わせの XNUMX つは、Apache Kafka と Spark Streaming のタンデムです。Kafka が受信メッセージ パケットのストリームを作成し、Spark Streaming がこれらのパケットを指定された時間間隔で処理します。

アプリケーションの耐障害性を高めるために、チェックポイントを使用します。 このメカニズムを使用すると、Spark ストリーミング エンジンが失われたデータを回復する必要がある場合、最後のチェックポイントに戻って、そこから計算を再開するだけで済みます。

開発したシステムのアーキテクチャ

使用したコンポーネント:

• アパッチカフカ は、分散型パブリッシュ/サブスクライブ メッセージング システムです。 オフラインとオンラインの両方のメッセージ消費に適しています。 データ損失を防ぐために、Kafka メッセージはディスクに保存され、クラスター内で複製されます。 Kafka システムは、ZooKeeper 同期サービス上に構築されています。
• Apache Spark ストリーミング - ストリーミング データを処理するための Spark コンポーネント。 Spark Streaming モジュールは、マイクロバッチ アーキテクチャを使用して構築されており、データ ストリームは小さなデータ パケットの連続シーケンスとして解釈されます。 Spark Streaming は、さまざまなソースからデータを取得し、それを小さなパッケージに結合します。 新しいパッケージは定期的に作成されます。 各時間間隔の開始時に新しいパケットが作成され、その間隔中に受信されたデータはすべてパケットに含まれます。 間隔の終わりに、パケットの増加は停止します。 間隔のサイズは、バッチ間隔と呼ばれるパラメータによって決まります。
• アパッチ スパーク SQL - リレーショナル処理と Spark 関数プログラミングを組み合わせます。 構造化データとは、スキーマ、つまりすべてのレコードに対して単一のフィールド セットを持つデータを意味します。 Spark SQL は、さまざまな構造化データ ソースからの入力をサポートしており、スキーマ情報が利用できるため、レコードの必要なフィールドのみを効率的に取得でき、DataFrame API も提供します。
• AWS RDS は比較的安価なクラウドベースのリレーショナル データベースであり、セットアップ、操作、拡張を簡素化する Web サービスであり、Amazon によって直接管理されます。

Kafka サーバーのインストールと実行

Kafka を直接使用する前に、Java があることを確認する必要があります。 JVM は作業に使用されます。

sudo apt-get update 
sudo apt-get install default-jre
java -version

Kafka を使用する新しいユーザーを作成しましょう。

sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo

次に、Apache Kafka の公式 Web サイトからディストリビューションをダウンロードします。

wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"

ダウンロードしたアーカイブを解凍します。

tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka

次のステップはオプションです。 実際のところ、デフォルト設定では Apache Kafka のすべての機能を完全に使用することはできません。 たとえば、メッセージを公開できるトピック、カテゴリ、グループを削除します。 これを変更するには、構成ファイルを編集しましょう。

vim ~/kafka/config/server.properties

ファイルの末尾に以下を追加します。

delete.topic.enable = true

Kafka サーバーを起動する前に、ZooKeeper サーバーを起動する必要があります。Kafka ディストリビューションに付属する補助スクリプトを使用します。

Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties

ZooKeeper が正常に起動したら、別のターミナルで Kafka サーバーを起動します。

bin/kafka-server-start.sh config/server.properties

「トランザクション」という新しいトピックを作成しましょう。

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transaction

必要な数のパーティションとレプリケーションを含むトピックが作成されていることを確認してください。

bin/kafka-topics.sh --describe --zookeeper localhost:2181

新しく作成されたトピックのプロデューサーとコンシューマーをテストする瞬間を見逃しましょう。 メッセージの送受信をテストする方法の詳細については、公式ドキュメントに記載されています。 メッセージを送信する。 さて、KafkaProducer API を使用して Python でプロデューサーを作成することに移ります。

プロデューサー執筆

プロデューサはランダム データ (毎秒 100 メッセージ) を生成します。 ランダム データとは、次の XNUMX つのフィールドで構成される辞書を意味します。

• ブランチ — 信用機関の販売時点情報管理の名前。
• 通貨 — 取引通貨;
• 単位 - 取引金額。 銀行による通貨の購入の場合、金額は正の数になり、販売の場合は負の数になります。

プロデューサーのコードは次のようになります。

from numpy.random import choice, randint

def get_random_value():
    new_dict = {}

    branch_list = ['Kazan', 'SPB', 'Novosibirsk', 'Surgut']
    currency_list = ['RUB', 'USD', 'EUR', 'GBP']

    new_dict['branch'] = choice(branch_list)
    new_dict['currency'] = choice(currency_list)
    new_dict['amount'] = randint(-100, 100)

    return new_dict

次に、send メソッドを使用して、サーバーに必要なトピックに JSON 形式でメッセージを送信します。

from kafka import KafkaProducer    

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                             value_serializer=lambda x:dumps(x).encode('utf-8'),
                             compression_type='gzip')
my_topic = 'transaction'
data = get_random_value()

try:
    future = producer.send(topic = my_topic, value = data)
    record_metadata = future.get(timeout=10)
    
    print('--> The message has been sent to a topic: 
            {}, partition: {}, offset: {}' 
            .format(record_metadata.topic,
                record_metadata.partition,
                record_metadata.offset ))   
                             
except Exception as e:
    print('--> It seems an Error occurred: {}'.format(e))

finally:
    producer.flush()

スクリプトを実行すると、ターミナルに次のメッセージが表示されます。

これは、すべてが希望どおりに機能することを意味します。プロデューサーがメッセージを生成し、必要なトピックに送信します。
次のステップでは、Spark をインストールし、このメッセージ ストリームを処理します。

Apache Spark のインストール

Apache Spark は、ユニバーサルで高性能なクラスター コンピューティング プラットフォームです。

Spark は、対話型クエリやストリーム処理など、幅広い種類の計算をサポートしながら、MapReduce モデルの一般的な実装よりも優れたパフォーマンスを発揮します。 大量のデータを処理する場合、速度は重要な役割を果たします。速度によって、数分から何時間も待つことなく対話的に作業できるようになります。 Spark を高速化する最大の強みの XNUMX つは、メモリ内で計算を実行できることです。

このフレームワークは Scala で書かれているため、最初にインストールする必要があります。

sudo apt-get install scala

公式 Web サイトから Spark ディストリビューションをダウンロードします。

wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"

アーカイブを解凍します。

sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/spark

Spark へのパスを bash ファイルに追加します。

vim ~/.bashrc

エディターを使用して次の行を追加します。

SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH

bashrc に変更を加えた後、以下のコマンドを実行します。

source ~/.bashrc

AWS PostgreSQL のデプロイ

残っているのは、ストリームから処理された情報をアップロードするデータベースをデプロイすることだけです。 このために、AWS RDS サービスを使用します。

AWS コンソール -> AWS RDS -> データベース -> データベースの作成に移動します。

PostgreSQL を選択し、「次へ」をクリックします。

なぜならこの例は教育目的のみであり、「少なくとも」無料サーバー (無料利用枠) を使用します。

次に、無料利用枠ブロックにチェックを入れます。その後、t2.micro クラスのインスタンスが自動的に提供されます。これは弱いですが、無料であり、私たちのタスクに非常に適しています。

次に、データベース インスタンスの名前、マスター ユーザーの名前、パスワードという非常に重要な項目が続きます。 インスタンスに myHabrTest、マスター ユーザーという名前を付けます。 ハーバー、パスワード: ハブ12345 そして、「次へ」ボタンをクリックします。

次のページには、外部からのデータベース サーバーへのアクセス可能性 (パブリック アクセシビリティ) とポートの可用性を制御するパラメータがあります。

VPC セキュリティ グループの新しい設定を作成しましょう。これにより、ポート 5432 (PostgreSQL) 経由でデータベース サーバーへの外部アクセスが許可されます。
別のブラウザ ウィンドウで AWS コンソールに移動し、[VPC ダッシュボード] -> [セキュリティ グループ] -> [セキュリティ グループの作成] セクションに進みます。

セキュリティ グループの名前 - PostgreSQL、説明を設定し、このグループを関連付ける必要がある VPC を指定して、[作成] ボタンをクリックします。

以下の図に示すように、新しく作成したグループのポート 5432 の受信ルールを入力します。 ポートを手動で指定することはできませんが、[タイプ] ドロップダウン リストから [PostgreSQL] を選択します。

厳密に言えば、値 ::/0 は、世界中からサーバーへの受信トラフィックが利用可能であることを意味します。これは標準的には完全に真実ではありませんが、例を分析するために、次のアプローチを使用できるようにしてみましょう。

ブラウザーのページに戻り、「詳細設定の構成」を開いて、VPC セキュリティ グループ セクションを選択します -> 既存の VPC セキュリティ グループを選択 -> PostgreSQL:

次に、データベースオプション -> データベース名 -> 名前を設定します - ハブDB.

バックアップ (バックアップ保持期間 - 0 日)、モニタリング、およびパフォーマンス インサイトの無効化を除き、残りのパラメーターはデフォルトのままにしておくことができます。 ボタンをクリックしてください データベースを作成する:

スレッドハンドラー

最終段階は Spark ジョブの開発です。Spark ジョブは、Kafka からの新しいデータを XNUMX 秒ごとに処理し、結果をデータベースに入力します。

上で述べたように、チェックポイントは SparkStreaming の中心的なメカニズムであり、フォールト トレランスを確保するために構成する必要があります。 チェックポイントを使用します。手順が失敗した場合、Spark Streaming モジュールは最後のチェックポイントに戻ってそこから計算を再開するだけで、失われたデータを回復できます。

チェックポイント機能は、チェックポイント情報が保存されるフォールト トレラントで信頼性の高いファイル システム (HDFS、S3 など) 上にディレクトリを設定することで有効にできます。 これは、たとえば以下を使用して行われます。

streamingContext.checkpoint(checkpointDirectory)

この例では、次のアプローチを使用します。つまり、checkpointDirectory が存在する場合、コンテキストはチェックポイント データから再作成されます。 ディレクトリが存在しない場合 (つまり、初めて実行された場合)、 functionToCreateContext が呼び出されて、新しいコンテキストが作成され、DStreams が構成されます。

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

KafkaUtils ライブラリの createDirectStream メソッドを使用して、「トランザクション」トピックに接続する DirectStream オブジェクトを作成します。

from pyspark.streaming.kafka import KafkaUtils
    
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 2)

broker_list = 'localhost:9092'
topic = 'transaction'

directKafkaStream = KafkaUtils.createDirectStream(ssc,
                                [topic],
                                {"metadata.broker.list": broker_list})

JSON 形式で受信データを解析します。

rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
                                       currency=w['currency'],
                                       amount=w['amount']))
                                       
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")

Spark SQL を使用して、単純なグループ化を実行し、結果をコンソールに表示します。

select 
    from_unixtime(unix_timestamp()) as curr_time,
    t.branch                        as branch_name,
    t.currency                      as currency_code,
    sum(amount)                     as batch_value
from treasury_stream t
group by
    t.branch,
    t.currency

クエリ テキストを取得し、Spark SQL を通じて実行します。

sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)

そして、結果として得られた集計データを AWS RDS のテーブルに保存します。 集計結果をデータベース テーブルに保存するには、DataFrame オブジェクトの write メソッドを使用します。

testResultDataFrame.write 
    .format("jdbc") 
    .mode("append") 
    .option("driver", 'org.postgresql.Driver') 
    .option("url","jdbc:postgresql://myhabrtest.ciny8bykwxeg.us-east-1.rds.amazonaws.com:5432/habrDB") 
    .option("dbtable", "transaction_flow") 
    .option("user", "habr") 
    .option("password", "habr12345") 
    .save()

AWS RDS への接続のセットアップについて少し説明します。 ユーザーとパスワードは「AWS PostgreSQL のデプロイ」ステップで作成しました。 [接続とセキュリティ] セクションに表示されるデータベース サーバー URL としてエンドポイントを使用する必要があります。

Spark と Kafka を正しく接続するには、アーティファクトを使用して smark-submit 経由でジョブを実行する必要があります。 スパークストリーミングカフカ0-8_2.11。 さらに、PostgreSQL データベースと対話するためのアーティファクトも使用し、 --packages 経由で転送します。

スクリプトの柔軟性を高めるために、メッセージ サーバーの名前とデータの受信元のトピックも入力パラメーターとして含めます。

それでは、システムを起動して機能を確認してみましょう。

spark-submit 
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2,
org.postgresql:postgresql:9.4.1207 
spark_job.py localhost:9092 transaction

すべてうまくいきました！ 以下の図からわかるように、StreamingContext オブジェクトの作成時にバッチ間隔を 2 秒に設定したため、アプリケーションの実行中は新しい集計結果が 2 秒ごとに出力されます。

次に、データベースに簡単なクエリを実行して、テーブル内のレコードの存在を確認します。 トランザクションフロー:

まとめ

この記事では、Spark Streaming を Apache Kafka および PostgreSQL と組み合わせて使用​​した情報のストリーム処理の例を検討しました。 さまざまなソースからのデータが増加するにつれて、ストリーミング アプリケーションやリアルタイム アプリケーションを作成するための Spark Streaming の実際的な価値を過大評価することは困難です。

完全なソース コードは私のリポジトリにあります。 GitHubの.

この記事について議論できることを嬉しく思います。皆さんのコメントを楽しみにしています。また、思いやりのある読者の皆様からの建設的な批判も期待しています。

私はあなたに成功を祈る！

psの。 当初はローカルの PostgreSQL データベースを使用する予定でしたが、AWS への愛着を考慮して、データベースをクラウドに移行することにしました。 このトピックに関する次の記事では、AWS Kinesis と AWS EMR を使用して、上記のシステム全体を AWS に実装する方法を説明します。 ニュースをフォローしてください！

出所： habr.com