スケーリング：JavaからScalaに切り替える理由

この記事では、Apache Hadoopでのデータ処理プロセスを説明するフレームワークであるTwitter Scaldingについて説明します。 Hadoopに加えて、フレームワークの歴史とともに、私は遠くから始めます。 次に、スケーリング機能の概要を説明します。 結論として、Javaを知っているがScalaにほとんど精通していない人には理解できるコード例を示します。

面白い？ 行こう！

MapReduceよりシンプル

MapReduceパラダイムが登場したばかりのとき、分散コンピューティングの開発を簡素化する画期的なステップでした。 ただし、マッパーとリデューサーを手動で記述するのは非常に面倒であることがすぐにわかりました。 開発をスピードアップするために、Map / Reduceの高レベルアドオン（Pig、Hive、Cascading、その他）が登場しました。 後者について説明しましょう。

カスケードは、データ処理プロセスを記述するためのJavaフレームワーク、いわゆる ワークフロー。 説明の後、CascadingはDBMSのクエリアナライザーのようなワークフローを分析し、一連のマップ/リデュースタスクの形式で実行計画を構築し、それらをHadoopクラスターに送信し、起動データと中間データを個別に管理します。 残念ながら、カスケードはかなり低レベルの抽象化で動作するため、人気の点では、他のデータ処理メカニズムに長い間失われています。

この状況から抜け出す良い方法が見つかりました。 Twitterは、Cascadingをニーズに合わせて調整し、その抽象化を従来のScalaツールでラップしました。 そのため、Scaldingが誕生しました-カスケードの上にScalaフレームワーク。 ここで余談をして、Scalaについて話すことができます。


そのため、Scalaの標準的なアプローチをワークフローデータ処理の記述に適用する試みは成功し、ScaldingはHive、Pig、およびそれらの多くの最新の対応物の人気に追いつく機会を得ました。 そのため、Scalaを学ぶことは理にかなっているので、今から行います。

スケーリングの概要

ここで、ScaldingとCascadingの内部構造に関連するすべてを意識的にスキップします。 これは、データを数える便利なインターフェイスを備えたブラックボックスであると想定しています。 すべてがうまくいけば、これらのフレームワークの内部構造に関する別の記事があります。

フラット操作

Scaldingの主な概念はPipeです。 パイプは、データがプログラマーに向かって流れるパイプラインです。 本質的に、これはJava 8のストリームに似ています。最初のPipe実装はタイピングをサポートしていませんでしたが、長続きしませんでした。 厳密な型指定のファンはTypedPipeを思い付きました-Javaの用語で、厳密に定義された型オブジェクト、ジェネリックを備えたパイプラインです。

TypedPipeの場合、 map 、 flatMap 、 filter 、 limitなど、いくつかの標準フロー操作が定義されています。 これらはすべてストリームに対するフラットな操作であり、理論的には、無制限の並列処理と任意の量のデータに対して効果的に実行できます。

TypedPipeのデータはどこかから読み取る必要があります。 このために、ScaldingにはSource （データソース）があります。 その唯一の目的は、PipeまたはTypedPipeを生成することです。 いくつかの既製のソースがあり、それらのほとんどはさまざまな形式のファイルから読み取りますが、任意のイテレーターから（したがってメモリー内のコレクションから）読み取る機能、そしてもちろん、ソースを判別する機能もあります。 重要なのは、同じカスケードとスケーリングのコードがHadoopクラスターとローカルデータの両方で機能することであり、これはテストに非常に便利です。

すべての操作が完了したら、データを保存します。 パイプラインの最後の部分であるシンクは、Scaldingでディスクに書き込む役割を果たします。 シンクはSourceに似ています。多くの場合、2つのインターフェイスを実装するのは同じクラスです。

グループ化操作

MapReduceを使用すると、TypedPipeで表されるストリームを再編成できます。 まず、SQLのGROUP BYに相当するキーでストリーム全体からレコードをグループ化するgroupByグループ化操作です。 グループ化した後、TypedPipe [V]は特別なフォームGrouped [ K、V]を取り、そこで追加の操作が利用可能になります。

最初に、 mapGroupメソッドとmapValuesStreamメソッドを使用して、グループ化が行われたキーKからペアとしてグループ化された[K、V]要素を取得し、このキーで発生したVのすべての値の反復子を取得できます。 Scalaのコレクション関数は、値反復子に適用されます。 しかし、通常、これは必須ではありません。 グループ化には、最も一般的なケースをカバーする多くのショートカット機能があります。

次に、GroupByはsortBy操作でソートできます。 その後、mapGroup、mapValuesStream、およびそれらのすべての派生物も適用できます。

第三に、グループ化された[K、V1]は別のグループ化された[K、V2]と結合できます。 ここでは、リレーショナルデータベースと同じルールが機能します。leftJoin 、 rightJoin 、 join （inner）、 outerJoin （full）が使用可能です。 出力はグループ化されます[K、（V1、V2）]。

グループ化されていないストリームにTypedPipe [（K、V）]ペアが含まれている場合、 hashJoin操作を適用できることに注意してください。 通常のGrouped.joinに似ていますが、メモリ内で実行されます。 これは、小さなディレクトリからのデータを充実させるのに適していますが、大きなテーブルではOOMにつながる可能性があります。

グループ化は、TypedPipe、キー、または値に対する操作を使用して、TypedPipeに戻すことができます。 最初はキーと値の両方を保存し、残りは1つのものを返します。

例によるスケーリング

ここで、フレームワークの主な機能を確認した後、例を使用してこれがどのように機能するかを見てみましょう。

私たちがRTBサイトであり、観察されたサイトのURLによるユーザーのクリックの履歴があるとします。 ストーリーは、URL、Timestamp、UserIdの3つの列を持つ巨大なTSVファイルで表示されます。

トピックに関するマークアップサイトもあります。 サイトの数はそれほど多くありませんが、最大数千です。 すべてのマークアップは、ドメインとトピックの列を持つ小さなTSVファイルに配置されます。

ユーザーがトピックを切り替える頻度を理解する必要があります。 これを行うには、ユーザーが1つのサブジェクトのサイトから別のサブジェクトのサイトに移動するときに、クリック履歴にそれらのイベントのみを残す必要があります。

この変換を行うコードを作成します。 発射インフラストラクチャは考慮されません。 興味があれば、完全なサンプルコードはgithubで入手できます。

Scalaでは、型のエイリアスを設定できます。 これは便利です 型宣言であるストリングと別のストリングを区別できるようになります。

 type Domain = String type UserId = String type Topic = String type Url = String type Timestamp = Long 

ドメインモデルからクラスを宣言します。

 case class Click(url: Url, ts: Timestamp, userId: UserId) case class SiteInfo(domain: Domain, topic: Topic) 

Scalaのケースクラスは、不変の値のクラスを記述する便利な方法です。 コンストラクター、ゲッター、およびその他の同様のコードが自動的に生成されます。

クリックして表を読む：

 val clicksPipe: TypedPipe[Click] = TypedPipe.from(TypedTsv[(Url, Timestamp, UserId)](pathToClicks)) .map(tuple => Click.tupled(tuple)) 

ここで、ソース（型（String、Long、UserId）の列を持つ型付きTSV）を発表しました。 次に、このソースをTypedPipeでラップしました。 さらに、便宜上、3列のタプル（Url、Timestamp、UserId）をClickクラスのオブジェクトに変換しました。

TypedPipe [クリック]が判明しました。

URLのドメインのみを残します。

 def url2domain(url: Url): Domain = { return new URL(url).getHost } val domainsPipe: TypedPipe[Click] = clicksPipe .map(click => click.copy(url = url2domain(click.url))) 

ドメインが件名で分割されているディレクトリを読み取り、すぐにhashJoinに適した形式にグループ化します。

 val sitesGroupByDomain: Grouped[Domain, SiteInfo] = TypedPipe.from(TypedTsv[(Domain, Topic)](pathToSites)) .map(tuple => SiteInfo.tupled(tuple)) .groupBy(siteInfo => siteInfo.domain) 

サイトのトピックに関するクリック情報の流れに追加します。 これを行うには、クリックストリームをドメインのディレクトリに参加させます。

 val clicksWithSiteInfo: TypedPipe[(Domain, (Click, SiteInfo))] = domainsPipe .map(click => (click.url, click)) .hashJoin(sitesGroupByDomain) 

クリックのストリームをユーザーごとにグループ化し、クリックのタイムスタンプで並べ替えます。 さらに、ドメインに関する情報にはもはや関心がなく、サイトの主題に関する情報だけで十分です。 これを行うために、トピックに対するユーザーの積極的な関心を一度に反映する補助クラスを導入します。

 case class TopicActivity(topic: Topic, ts: Timestamp, userId: UserId) val topicActivityStreamPerUser: SortedGrouped[UserId, TopicActivity] = clicksWithSiteInfo .map(tuple => { val (domain, (click, siteInfo)) = tuple TopicActivity(siteInfo.topic, click.ts, click.userId) }) .groupBy(activity => activity.userId) .sortBy(activity => activity.ts) 

最も難しい瞬間-ユーザーアクティビティの流れの中で、トピックを切り替える瞬間をキャッチする必要があります。 切り替えをキャッチするために、ScalaでJavaスタイルの関数を作成します。 結果はArrayBuffer（ArrayListの類似物）に蓄積され、非常に長いストーリーでOOMにつながる可能性があります。

 def topicSwitches(userId: UserId, activities: Iterator[TopicActivity]): Iterator[TopicActivity] = { val result = ArrayBuffer[TopicActivity]() var firstTs = 0l var lastTopic = null.asInstanceOf[Topic] for (activity <- activities) { if (firstTs == 0l || lastTopic != activity.topic) { result.append(activity) firstTs = activity.ts lastTopic = activity.topic } } result.toIterator } val firstTopicActivitiesPipe: TypedPipe[TopicActivity] = topicActivityStreamPerUser .mapGroup((userId, activities) => topicSwitches(userId, activities)) .values 

各関心の最初のアクティビティのみがストリームに残りました。 これらは、ユーザーの関心の焦点が時間とともにどのように変化したかを追跡するために使用できます。 結果をファイルに書き込むことは残っています。

 firstTopicActivitiesPipe .map(activity => (activity.topic, activity.ts, activity.userId)) .write(TypedTsv(args.required("output"))) 

以上です。 文字通り40行で、重要なデータ変換を説明しました。

スカラウェイの最終コード

正規のスカラーウェイに従うと、コードはさらに短くなります。 また、トピックを反復アプローチから機能アプローチに切り替えるための検索機能を書き換えて、バッファーの使用を削除することができます。 今、プロセスは無限の入り口でも落ちません。 理論的には...

 def topicSwitches(userId: UserId, activities: Iterator[TopicActivity]): Iterator[TopicActivity] = { activities.scanLeft(Helper())((helper, activity) => { if (helper.topic.isEmpty || helper.topic.get != activity.topic) { Helper(Some(activity.topic), activity.ts, true) } else { Helper(helper.topic, helper.firstTs, false) } }).filter(_.firstVisit).map(helper => TopicActivity(helper.topic.get, helper.firstTs, userId)) } TypedPipe.from(TypedTsv[(Url, Timestamp, UserId)](pathToClicks)) .map(tuple => Click.tupled(tuple)) .map(click => click.copy(url = new URL(click.url).getHost)) .map(click => (click.url, click)) .hashJoin( TypedPipe.from(TypedTsv[(Domain, Topic)](pathToSites)) .map(tuple => SiteInfo.tupled(tuple)) .groupBy(_.domain) ) .map({case (_, (click, siteInfo)) => TopicActivity(siteInfo.topic, click.ts, click.userId)}) .groupBy(_.userId) .sortBy(_.ts) .mapGroup(topicSwitches) .values .write(TypedTsv(outputPath)) 

次の記事では、インラインデータの処理とテストのためにコードを整理する問題について説明します。 そして最後に、すべてが内部からどのように機能するかを説明します。

---

資源

完全なgithubのサンプルコード
スケーリングwiki
ブック「Scaldingを使用したMapReduceのプログラミング」