Apache Spark

このポストについて データ基盤移行について書いていくシリーズです。 シリーズ一覧はこちらから。 前回 Part 6. メタデータ編ではメタデータの運用開始と移行について書きました。 今回は最も苦労してようやく最近終わった ELT 処理の移行について記載します。 この移行については Part 3. アーキテクチャ編でも触れているのですが、今回はもう少し詳細に書きます。 ELT 処理の移行の概要 このデータ基盤移行では Amazon Athena を中心とした構成の旧データ基盤から、Databricks で構成された新データ基盤への移行を行いました。 ELT 処理まわりについて、移行前は Glue Job (PySpark) と Athena クエリを Apache Airflow から回すというような形になっていました。 移行後は処理エンジンは Databricks SQL に統一し、それを dbt から回すという形にしました。 なぜこうしたのかについては Part 3. アーキテクチャ編をご覧ください。 一方でそっちには書いていなかったのですが、table 間の依存関係の管理を Apache Airflow から dbt に変更することによってかなり開発・運用の手間が減ったという面があります。 Aiflow はいろいろなことができ、また UI も使いやすく worfklow orchestration の仕組みとして悪いものではありません。 しかし DAG 上で task 間の関係として依存関係を管理するのはそれなりに手間がかかります。 dbt の場合は下流 model の定義において {{ ref('upstream_model') }} のように上流 model 名を指定して参照するだけで依存関係を表現できます。 これはかなり楽でした。 ...

ほとんどバッチ処理しか書いたことのない者だがストリーム処理のシステムを開発することになった。 それにあたって独学で調べたことなどまとめておく。 ストリーム処理とは そもそも “ストリーム処理” とは何を指しているのか。 以下の引用が簡潔に示している。 a type of data processing engine that is designed with infinite data sets in mind. Nothing more. – Streaming 101: The world beyond batch こちらは “streaming system” について述べたものだが、つまり終わりのないデータを扱うのがストリーム処理ということである。 例えば web サービスから生まれ続けるユーザ行動ログを逐次的に処理するというのがストリーム処理。 web サービスが終了しないかぎりはユーザ行動ログの生成には終わりがない。 これに対して “1日分のユーザ行動ログ” 等のように有限の量のデータを切り出して処理する場合、これはバッチ処理となる。 ストリーム処理とバッチ処理の違いは扱うデータが無限なのか有限なのかということだ。 この後触れていくが、この終わりのないデータを継続的に処理し続けるというところにバッチ処理にはない難しさがある。 なぜストリーム処理なのか なぜストリーム処理なのか。 ひとえに逐次的な入力データに対する迅速なフィードバックが求められているからと言えるだろう。 迅速なフィードバックがビジネス上のメリットとなることは自明だ。 SNS の配信 カーシェアリングにおける配車や料金設定 クレジットカードや広告クリックなどの不正検知 もしこれらの application が例えば hourly のバッチ処理で実装されていたらどうだろうか。 まあ待っていられない。 ...

2020-07-31 にオンライン開催された Spark Meetup Tokyo #3 Online に参加した。 感想などメモしておく。 全体感 トピックとしては主に Spark 3.0 Spark + AI Summit 2020 Spark 周辺要素 といったところだろうか。 最近のコミュニティの動向や関心を日本語で聞くことができてよかった。 運営 & スピーカーの皆様、ありがとうございます。 発表 発表資料は公開されたら追加していく。 SPARK+AI Summit 2020 イベント概要 スピーカー: @tokyodataguy さん Summit は金融業界の参加者が増えているらしい Spark で最も使われている言語は Python とのことだったが、Databricks の notebook サービスの話だった プロダクションコードではまた違うのだろう Spark 3.0 の update をざっくりと Spark 周辺要素の話をざっくりと Koalas Delta Lake Redash MLflow Introducing Koalas 1.0 Introducing Koalas 1.0 (and 1.1) from Takuya UESHIN スピーカー: @ueshin さん ...

Spark バッチ処理の問題を調べていたら分離レベルという概念にたどりついた。 分離レベルについて調べたので、Spark の問題の内容と絡めて記しておく。 考えてみれば当たり前でたいした話ではない。 分離レベルとは トランザクションの挙動についての暗黙の理解 アドホックな分析クエリやプロダクションコード中のクエリを書くとき、その単一のクエリのトランザクションにおいて「同時に実行されている別のクエリの commit 前の状態や commit 結果に影響され、このクエリの結果がおかしくなるかもしれない」ということは通常考えない。 トランザクションはデータベースのある時点の状態に対して正しく処理される、というほぼ無意識の理解をおそらくほとんどの開発者が持っている。 多くの場合この理解は間違っていない。 それはなぜかというと DB 等のデータ処理フレームワークがある強さの分離レベルを提供しているからである。 いろいろな分離レベル ACID 特性のうちの1つ、分離性 (Isolation) の程度を表すのが分離レベル。 トランザクション中に行われる操作の過程が他の操作から隠蔽されることを指し、日本語では分離性、独立性または隔離性ともいう。より形式的には、独立性とはトランザクション履歴が直列化されていることと言える。この性質と性能はトレードオフの関係にあるため、一般的にはこの性質の一部を緩和して実装される場合が多い。 – Wikipedia ACID (コンピュータ科学) 分離レベルには名前のついたものがいくつかあり、分離性の保証の強さが異なる。 具体的にはトランザクションの並行性の問題への対応力が異なる。 名著「データ指向アプリケーションデザイン」の第7章で分離レベルについて詳しく述べられているので、以下ではそちらからの引用。 分離レベルを弱い順に並べる。 read uncommitted このレベルではダーティライトは生じませんが、ダーティリードは妨げられません。 read committed データベースからの読み取りを行った際に見えるデータは、コミットされたもののみであること（ダーティリードは生じない）。 データベースへの書き込みを行う場合、上書きするのはコミットされたデータのみであること（ダーティライトは生じない）。 snapshot isolation スナップショット分離の考え方は、それぞれのトランザクションがデータベースの一貫性のあるスナップショットから読み取りを行うというものです。すなわち、トランザクションが読み取るデータは、すべてそのトランザクションの開始時点のデータベースにコミット済みのものだけということです。 serializability この分離レベルはトランザクションが並行して実行されていても、最終的な答えはそれぞれが1つずつ順番に、並行ではなく実行された場合と同じになることを保証します。 日本語で「分離レベル」を検索すると snapshot isolation の代わりに repeatable read が出てくる事が多い。 しかし repeatable read の名前は実装によって意味が違っていたりして扱いが難しいらしい。 ...

はじめに Apache Spark 3.0.0 がリリースされました。 Spark Release 3.0.0 release note を見て個人的に気になったところなど簡単に調べました。 書いてみると Databricks の記事へのリンクばっかになってしまった… 全体感 こちらの記事を読めば全体感は OK. Introducing Apache Spark 3.0 公式の release note には Python is now the most widely used language on Spark. とあってそうなん？ってなったけど、こちらの記事だと Python is now the most widely used language on Spark and, consequently, was a key focus area of Spark 3.0 development. 68% of notebook commands on Databricks are in Python. と書いてありどうやら Databricks の notebook の話らしく、だったらまあそうかなという感じ。 プロダクトコードへの実装というよりは、アドホック分析や検証用途の話なんでしょう。 ...