背景

この記事では、ストリーム処理に関する内容をまとめています。

そもそもストリーム処理はどういう用途で使用されているの?

無限に発生し続けるデータを処理する場合
処理が永続的に継続する場合

例えば、Beaconなどから検知して送られてくる位置情報のストリームデータがあれば、仮に対象範囲外の位置データは必要とせず、処理しないとする場合に、必要なデータのみを保管し、それ以外のデータは価値がないのでフィルタリングするケースがあると思います。そのような時に、無限かつ流動的に発生するデータをリアルタイムで処理する為にストリーム処理を使用します。

課題を解決するために

Akka Actor

リアルタイムの処理をするための一つの解決策として、Akka Actorが挙げられます。
(Akka Actorに関しては、次回の記事で詳細にまとめていきます。)

What is Akka Actor

アクターと呼ばれるメッセージの送受信出来るオブジェクトが、非同期でお互いにやりとりするという仕組みです。しかし、Actor間のメッセージングを安定して実現する事は難しいです。

その一つの要因として

Publisher側の処理が早く完了する場合に、Subscriber側のバッファーが溢れてしまう。
Subscriberに対して、Publisher側の処理を抑えた場合は、その分だけ無駄になる

Akka Stream

そこで、Akka Streamを提案します。主な理由としては、

Back Pressureによりバッファ溢れせず、性能を最大限まで引き出す事ができる。
構成要素のInput/Outputに型がついている
マイクロサービス化の実現

Back Pressureとは

Subscriberが自分が処理できる量をPublisherにリクエストを送ることで、無駄なくSubscriberが処理できる量を処理する仕組み。

Subscriberが自分が処理できる量をPublishserに伝え、PublisherはSubscriberから伝えられた情報に合わせて、適切な量のメッセージをSubscriber側に送り、これによってSubscriberを溢れさせず、Subscriberのリソースを最大限に使って動作させることが可能となります。
BackPressure自体はakka-streamが勝手に裏側で制御してくれているので、特に制御について意識する必要はないです。

実際にakka streamsを実装してみた

Source

データの出力を行う。アウトプットチャンネルをひとつだけ持つ。(Publisher)

Flow

インプットとアウトプットのチャンネルをひとつずつ持つ。Sourceで取得したデータをフィルタリングの処理を実装したりする。

Sink

データの受信側。(Subscriber) インプットのチャンネルをひとつだけ持つ。

RunnableGraph

Source + Flow + Sink　の要素を３つを連結したグラフをRunnableGraphと言います。
データを処理する一連の流れを形態を表す物になります。

ソースコード

フローの流れ
- webページのスクレイピング →
  保存した画像サイズフィルター → 1000*1000以下のimgファイル削除

class ScrapeStream extends ImgHelper{

  implicit val system = ActorSystem("ScrapeSystem")
  implicit val materializer = ActorMaterializer()
  implicit val ec: ExecutionContext = ExecutionContext.Implicits.global

  val targetUrls: List[String] = List (
    // スクレーピングしたいページのURL
    "http://hoge.com"
  )

  //webページのスクレイピング
  val source = Source(targetUrls).mapAsyncUnordered(3) { url => scrape(url)}

  //imgサイズフィルター
  val step = Flow[List[File]].mapAsyncUnordered(3) { files =>
    Future(for {
        file <- files
        if (check(file))} yield file)
  }

  //imgファイルの移動
  val sink = Sink.foreachParallel[List[File]](3) { files => files.map(file => delete(file))}

  val graph = source via step to sink
  
  def executeScrape(): Unit = {
    graph.run()
  }
}

trait ImgHelper {

def delete(file: File) = {
 file.delete()
}

def scrape(url: String) = Future {
  val img = JsoupBrowser().get(url) >?> elementList("img")
  img match {
    case Some(l) => for {e <- l} yield {save(e.attr("src"))}
    case None => List.empty[File]
  }
}

private def save(url: String) = {
  val file = new File("保存したいファイルパス")
  Resource.fromFile(file).write(Resource.fromURL(url).byteArray)
  file
}

def check(file: File) = {
  val image = ImageIO.read(file)
  image.getHeight() <= 1000 && image.getWidth() <= 1000
}

まとめ

Source、Flow、Sinkの独立した要素として実装することで、各々がマイクロサービス化されていることもメリットの一つかと思います。
お互いのフローに影響を与えずに、Sourceを「S3からのファイル読み込み」に書き換えることや、Sinkの部分を「別サーバへのファイルバックアップ処理」などに変更することも出来ます。
それぞれの要素のInput/Outputの型を合わせるだけで、処理要素を追加したり、入れ替えたりすることが容易に書き換えることが出来るのも、akka streamsのメリットかと思います。