初めまして、カケハシのデータ基盤チームでデータエンジニアをしている伊藤と申します。 最近の悩みは、二郎ラーメンを食べていないのに「二郎ラーメンの匂い(臭い？)がする」と同居人に言われることです。私のニュースは置いといて、カケハシでは全社的なデータ活用基盤のプラットフォームとしてDatabricksを採用してから半年以上経過しました。 Databricksを導入してから今まではバッチ処理しかしてませんでしたが、最近になってAutoLoaderを利用してストリーム処理をするようになりました。その対象として、弊社が提供している薬歴システムのMusubiの監査ログを扱ったので紹介させて頂きます。 Au…

お金は好きですか？ コストを削減したいみなさん、ようこそ。 原価を低減したいみなさん、ようこそ。 サーバレスのビジネスでの優位性は多数ありますが、今回はその中でもコスト最適化の面から説明します。 やればやるだけ成果が見える: 努力が必ず報われる物語 レイテンシを1ms削るようなリリースができたとしましょう。(ネットワーク系のコストは同等と仮定します) 従来の開発では、CPUやメモリ利用量が若干減った程度では、コストは変わりませんでした。利用率が100%にならないようバッファを持たせており、多少削ってもサーバーの台数削減につながらないからです。あえてリソースを減らすとしても、利用率100%のリス…

初めまして、カケハシのデータ基盤チームでデータエンジニアをしている伊藤と申します。 最近の悩みは、二郎ラーメンを食べていないのに「二郎ラーメンの匂い(臭い？)がする」と同居人に言われることです。私のニュースは置いといて、カケハシでは 全社的なデータ活用基盤のプラットフォームとしてDatabricksを採用し てから半年以上経過しました。 Databricksを導入してから今まではバッチ処理しかしてませんでしたが、最近になってAutoLoaderを利用してストリーム処理をするようになりました。その対象として、弊社が提供している薬歴システムのMusubiの監査ログを扱ったので紹介させて頂きます。 AutoLoaderとは Databricks上で特別な設定無しで、クラウドストレージに到着した新しいデータを段階的かつ効率的にDelta Lakeに取り込む仕組みです。AutoLoaderにはディレクトリごとに数百万のファイルにまで拡張できるスケーラビリティの対応や導入が容易で使いやすいといったメリットがあります。 以前、spark streamingでストリーム処理を行い、Delta Lakeに取り込んた際の課題として、読み込み対象のスキーマが想定外に更新されてデータ取得によく失敗していました。AutoLoaderにはスキーマの変更が起きた際に通知を行い、無視するか失われるデータを救助できる仕組みがあるため、その課題を解決できます。 AutoLoaderの動作原理 AutoLoaderでは新規ファイルを検知するために ディレクトリ一覧モード と ファイル通知モード があります。 本記事で紹介する事例では、Kinesis Firehoseを使用してファイルを日付順にアップロードしており、ディレクトリ一覧モードでAutoLoaderを起動し、API呼び出し数を削減しております。 アーキテクチャ Musubiの監査ログはCloudWatchに出力されています。AWSのKinesisを利用してCloudWatchからJSONデータをS3に格納しています。AutoLoaderで監査ログを処理する前は、日次でバッチ処理を行っていたので以下のようなアーキテクチャとなっていました。 最近になって監査ログデータのリアルタイム性が求められてきました。今までのアーキテクチャだとデータ抽出まではリアルタイムになっていたのですが、Databricksに読み込む際にバッチ処理をしていたため、利用者にとってはリアルタイム性に欠けていました。そのためアーキテクチャの見直しを行い、日次バッチからAutoLoaderで処理を行うよう変更しました。 Databricksでは、増分データの取り込みにはDelta Live Tables(DLT)でAutoLoaderを使用することが推奨されていますが、Unity Catalog/DLTの統合は本記事執筆時点の2023年1月31日時点ではサポートされていないため、現時点ではDelta Live Tablesを使用していません。 AutoLoaderでJSONファイルの読み込み 以下を実行することでS3に到着するJSONが順次処理されるストリームが起動します。 df = spark.readStream.format( "cloudFiles" ) \ .option( "cloudFiles.format" , "json" ) \ .option( "cloudFiles.schemaLocation" , "checkpointに使用するpath" ) \ .option( "cloudFiles.schemaHints" , "type string, target string, action string, year string, month string, day string, hour string) \ .option(" cloudFiles.partitionColumns ", " year,month,day,ho ur") \ .option(" cloudFiles.useIncrementalListing ", True) \ .option(" lineSep ", " \n ") \ .load(" 取り込みデータのpath ") .format("cloudFiles") と指定することでAutoLoaderを利用しています。\ .option("cloudFiles.schemaLocation") にチェックポイントディレクトリを指定することで、AutoLoaderがどこまでファイルを処理したか記録しています。\ 予めスキーマ情報がわかっていたので .option("cloudFiles.schemaHints" にカラム名と型を指定することでスキーマ情報を強制させています。\ Kinesisで順次パーティション区切りでストリーミングデータが送られてくるので、 .option("cloudFiles.partitionColumns") に年/月/日/時を指定しています。\ AutoLoaderの動作原理で触れましたがディレクトリ一覧のモードを使用しており、日付順でS3にファイルがアップロードされるため、 .option("cloudFiles.useIncrementalListing", True) を指定して、ディレクトリ一を完全に一覧するのではなく、インクリメンタルな一覧を適用することにより新規ファイルを検知するのに必要なAPI呼び出し数を削減しています。\ .option("lineSep) には二つの連続するJSONレコードの区切り文字として\nを指定しています。 Databricks Unity Catalogに書き込み 以下を実行することで読み込んだストリームをDatabricksのUnity Catalogに書き込みます。 df = spark.writeStream.format( "delta" ) \ .option( "checkpointLocation" , "checkpointに使用するpath" ) \ .outputMode( "append" ) \ .option( "mergeSchema" , "true" ) \ .partitionBy( "year" , "month" , "day" , "hour" ) \ .trigger(processingTime= "1 minutes" ) \ .toTable( "unity catalogに保存するテーブル名" ) .option("mergeSchema", "true") で新規カラムが追加された場合に自動でDeltaに取り込むようオプション指定しています。\ .trigger(processingTime="1 minutes") に1 minutesを指定することで、1分間隔でJSONを順次処理しています。 JSONファイル処理部分で躓いた ストリーミングでデータを処理すると、スキーマの予期しない変更や取り込み対象ファイルのデータが不正といった様々な要因でエラーに遭遇するかと思います。今回AutoLoaderでJSONファイルを処理する際に以下のエラーに遭遇しました。 org.apache.spark.sql.catalyst.util.UnknownFieldException: Encountered unknown field(s) during parsing: {} 上記のエラー原因は、データスキーマ変更が要因でない場合、データの中に空のJSONが存在すること起因の不具合によるものです。対処方法としては、以下２パターンのどちらかになるかと思います。 取り込み対象データの空のJSONを削除する エラー後、AutoLoaderを再実行 今回はAutoLoaderを再実行することで問題なくデータを取り込むことができて、データ利用者に最新の情報を提供することができました。 まとめ 以上、DatabricksのAutoLoaderを紹介させて頂きました。簡単にセットアップができて最新の情報をいち早くデータ利用者に届けられる仕組みは大変素敵なものだと実感しました。 本記事ではDelta Live Tablesについて触れませんでしたが、Unity Catalog/Delta Live Tablesの統合が正式版となった暁には社内でDelta Live Tables(DLT)でAutoLoaderを使用する動きが加速するかと思います。 一緒に手を動かしてデータ基盤構築しませんか。少しでも興味を持ってくださった方がいらっしゃれば以下からご応募お待ちしております。

KAKEHASHI でバックエンドエンジニアをしている横田です。 私が運用している Web サービスでは、AWS Glue で ETL 処理をしたデータを Aurora MySQL に投入することでユーザーが利用できるようにしています。 その中でも「データを Aurora MySQL に投入する」方法に関して、今まで色々なパターンを試してきました。 AWS Glue の Job で作成したデータを Aurora に投入するいくつかのパターンとそのメリット・デメリットについて紹介できればと思います。 Aurora MySQL にロードする 3 つのパターン データを MySQL に insert…

はいさい！カケハシの新米メンバー、オースティンと申します。 沖縄から参上しております！ 概要 RxJS の mergeMap と switchMap の違いと使い方について解説します。 背景 Observable を使っていると、必ず直面する問題があります。それは、 複数の Observable をどうやって一緒に実行できるか 、という問題です。 とある Observable の処理が終わった後に、そのデータを元に、別の Observable でさらに非同期処理をすることは開発者として多々あります。 Promise の then でまた別の Promise を返してチェーンしていくのと同じことです。 しかし、RxJS では、 Promise と違って Observable を直列にチェーンするのに使う pipe の OperatorFunction が複数存在します。代表的なのは、 mergeMap 、 mergeWith 、および switchMap です。 なぜ RxJS には複数のチェーン手法が存在するのか この疑問は湧きませんか？ この疑問に答えるためにはまず、 Observable と Promise はどう異なるのか考える必要があります。 筆者に言わせれば、 Observable の最も大きな違いは、 未解決の Observable の処理を取り消す、つまり止めることが簡単にできる点 でしょう。 Promise では、 Promise が解決してもその中で実行された処理を取り消す方法がありません。以下のコードを見ましょう。 const promise = new Promise((resolve) => { let count = 0; setInterval(() => console.log( `Loading ${++count} seconds.` ), 1000); setTimeout(resolve, 5000); } ); promise.then(() => console.log( "Resolved!!" )); 読者は読んでお分かりかと思いますが、このコードを実行してみると、以下のように、 Promise が解決された後でも、ログが出続けるのです。 同じ例を Observable で書きます。 import { Observable } from "rxjs" ; const observable$ = new Observable((observer) => { let count = 0; const interval = setInterval(() => console.log( `Loading ${++count} seconds.` ), 1000); setTimeout(() => { observer.next( "Timeout ended." ); observer.complete(); } , 5000); return () => clearInterval(interval); } ); observable$.subscribe( { next: console.log, complete: () => console.log( "Completed!" ) } ); Observable の executor 関数が、戻り値としてその interval をクリアする関数を返すのですが、これが Observable が解決（complete）された時に呼ばれるのです。 すると、 Promise と違って永遠に続く interval が残されません。 この 後片付け の機能が、 Observable の非常に優れたところなのです 。 そして、本記事につながる部分ですが、この後片付けがあるからこそ、 Observable をチェーンする時の手法が用途によって異なるのです。 mergeMap とは何か mergeMap は、map と似ていますが、簡単にいうと、一つの Oberservable から流れたデータを違う Observable に流すチェーン OperatorFunction なのです。 以下の例を見てみましょう。 import { fromEvent, scan, mergeMap, interval, takeWhile, tap, combineLatest, of } from "rxjs" ; const docClick$ = fromEvent( document , "click" ); const clickCount$ = docClick$.pipe( scan((acc) => acc + 1, 0), tap((count) => console.log( `Document was clicked ${count} time(s).` )) ); clickCount$ .pipe( mergeMap((clickCount) => { const intervalPerClickCount = [ of(clickCount), interval(500).pipe(takeWhile((i) => i < 5)) ] ; return combineLatest(intervalPerClickCount); } ) ) .subscribe(( [ clickCount, i ] ) => console.log( `mergeMap: Click no ${clickCount} , interval count: ${i} ` )); document をクリックすると、クリックした回数をまず clickCount$ で足し算します。 それから、 mergeMap を使って新しい Observable を返します。 その新しい Observable は、 combineLatest で合わせた二つの Observable 、 of(clickCount) と interval です。 要するに、 mergeMap で Observable<number> を Observable<[number, number]> というふうに変えています。 試してみる document を一回クリックすると以下のように出力されます。 Document was clicked 1 time(s). mergeMap: Click no 1, interval count: 0 mergeMap: Click no 1, interval count: 1 mergeMap: Click no 1, interval count: 2 mergeMap: Click no 1, interval count: 3 mergeMap: Click no 1, interval count: 4 document を 2 回連発でクリックすると以下のようにログが出力されます。 Document was clicked 1 time(s). Document was clicked 2 time(s). mergeMap: Click no 1, interval count: 0 mergeMap: Click no 2, interval count: 0 mergeMap: Click no 1, interval count: 1 mergeMap: Click no 2, interval count: 1 mergeMap: Click no 1, interval count: 2 mergeMap: Click no 2, interval count: 2 mergeMap: Click no 1, interval count: 3 mergeMap: Click no 2, interval count: 3 mergeMap: Click no 1, interval count: 4 mergeMap: Click no 2, interval count: 4 ここで重要な観察ですが、 2 回目のクリックがあっても、1 回目のクリックを元にした combineLatest の Observable<[number, number]> は最後まで続くのだ という結果を記憶に留めておきましょう。 switchMap switchMap は mergeMap と同じように、上流の Observable を新しい Observable に合わせます。 しかし、重要な違いがあります。 switchMap は、上流の最も最新も値をのみとって、下流の新しい Observable に流すのです 。 たとえ、以前の値に基づいて流した下流の Observable がまだ解決されていないとしても、 それらの Observable を止めるのです 。 上記のソースコードで mergeMap が使われていたところを switchMap にしてみましょう。 clickCount$ .pipe( switchMap((clickCount) => { const intervalPerClickCount = [ of(clickCount), interval(500).pipe(takeWhile((i) => i < 5)) ] ; return combineLatest(intervalPerClickCount); } ) ) .subscribe(( [ clickCount, i ] ) => console.log( `switchMap: Click no ${clickCount} , interval count: ${i} ` )); これも実験してみましょう！ 試してみる 1 回だけクリックしてみる Document was clicked 1 time(s). switchMap: Click no 1, interval count: 0 switchMap: Click no 1, interval count: 1 switchMap: Click no 1, interval count: 2 switchMap: Click no 1, interval count: 3 switchMap: Click no 1, interval count: 4 mergeMap と同じ結果です。 2 回連発してクリックしてみる Document was clicked 1 time(s). Document was clicked 2 time(s). switchMap: Click no 2, interval count: 0 switchMap: Click no 2, interval count: 1 switchMap: Click no 2, interval count: 2 switchMap: Click no 2, interval count: 3 switchMap: Click no 2, interval count: 4 なるほど！**最後のクリックだけ、5 回の interval が出たのです！ まとめ ここまで mergeMap と switchMap の違いを解説してきましたが、いかがでしょうか？ 似たような効果があるのに、歴然な違いがあるので、筆者は知った時に驚きました。 この違いを知らずにコードを書いていると、解せないエラーが起きそうな気がします。 たとえば、HTTP リクエストでクリックに対してログを記録させたい時に、どれを使えばいいと思いますか？ ビジネスモデルにもよるのですが、 switchMap だと、ログのリクエストがまだ終わっていないのに、ユーザーがまたクリックすると、前のログのリクエストがキャンセルされ、 最後のクリックだけがログに残る結果になるのです 。なので、 mergeMap が向いているでしょう。 逆に、自動推測などだと、最新の値だけに対してリクエストを投げたいはずなので、 mergeMap より switchMap が適しているのではないでしょうか？ RxJS は奥深くて強力なのですが、強力だからこそ誤った使い方をすると痛い目に遭うなと思っています。 どんどん理解を深めていきましょう！

こんにちは！カケハシにて薬局と患者の関係性を向上させるためのツールである 患者リスト というWEB業務アプリケーションを開発している小室と申します。 本プロダクトのフロントエンドの開発環境としては、React + esbuildを採用しており、採用の経緯や実践している環境構築方法などは以下の通り、TechPlayやQiitaなどに記事を投稿してきました。 TechPlay: 新規事業プロダクト開発時の技術選定どうやった？ スライド Qiita: esbuild + React(TS) で実現する超軽量な開発環境 しかしながら、esbuildは標準でsassに対応しておらず、今までの環境ではCS…

こちらの記事は、カケハシ Advent Calendar 2022の25日目の記事になります。 こんにちは、四番隊隊長とは声が低いこと以外何一つ共通点がないCTOの海老原です。 すみません、タイトルは釣りタイトルです。何故こんな釣りをアドベントカレンダーのラストに持ってきたかですが…しばしばスタートアップの経営陣の最大のミッション・役割の一つとして採用が挙げられますが、ご多分に漏れず私も日常の時間のかなり多くの部分をソフトウェアエンジニアやデザイナー、プロダクトマネージャーのような開発系職種のカジュアル面談での事業説明やオファー面談に割いています。 その中でよく候補者の方からご質問を頂くわけで…

こちらの記事はDatabricks Advent Calendar 2022の25日目の記事になります。 こんにちは、カケハシでMusubi Insightという薬局向けBIツールのバックエンドエンジニアをしている高田と申します。 BIツールを開発しているということもあり日常的にETL処理の実装を行っていますが、普段の開発ではAWS Glueを採用しています。 しかし、カケハシでは全社的なデータ活用基盤のプラットフォームとしてDatabricksが採用されたこともあり、とあるプロジェクトでDatabricksを活用してデータエンジニアやデータアナリストとうまく協業できたので、その時の事例を紹介…

こちらの記事はDatabricks Advent Calendar 2022の24日目の記事です。 はじめに 初めまして。カケハシでデータサイエンティストをしている赤池です。 弊社はフルリモートで業務できるため今年9月から地元の仙台市で業務していますが、本格的な冬の到来を前に戦々恐々しています。（寒い。雪。路面凍結。） さて、あなたは「Pandas API on Spark」を知っていますか？ これは「pandasと同じ書き方でSpark上で処理を実行できる」という代物で、pandasでは処理に時間がかかる or そもそも扱えないような大規模データを、ほとんどpandasと同じ感覚で処理できる…

この記事は、カケハシ Advent Calendar 2022 の 24 日目 の記事になります。 こんにちは、木村です。(@kimutyam) 医薬品発注管理最適化領域の新規事業のテックリード兼エリアPO、プラットフォームドメイン全体のアーキテクト、データ基盤チームのアーキテクトサポートの3つを兼務しているエンジニアです。直近はプラットフォームドメインの領域のテナント管理の立ち上げに集中をしています。 プラットフォームドメインとは カケハシはMusubiを始めとした5つのサービスを提供しています。これらは独立して価値を届けられる単位ではありますが、サービスを横断してしなやかなユーザー体験を提…

こちらの記事はDatabricks Advent Calendar 2022の23日目の記事です。 はじめに 初めまして。カケハシにてデータサイエンティストをしている赤池です。 業種的に、自己紹介の際に統計学のビッグネームとの関係性を聞かれることがたまにありますが全く関係ありません。統計学もがんばります。 突然ですが、あなたの分析環境では「DBから抽出したデータをPythonやRなどで利用する際にうまく連携できていない」なんてことはありませんか？ そして「この処理だけはRでやりたいが、そのためだけに別環境を開きたくない」なんて思うことはありませんか？ このブログでは、データサイエンティストの立…

こちらの記事はDatabricks Advent Calendar 2022の23日目の記事です。 カケハシのデータ基盤チームの松田です。カケハシでは今年の7月からDatabricksを利用しており、そろそろ半年ぐらい経とうとしています。Databricksを採用した背景については、以下の記事に詳細をまとめていますので、まだ拝見されていない方は見ていただければ幸いです。色々な人から「とても参考になった！」と反響を呼んでいるオススメの記事です！ 今回はDatabricks導入のタイミングで、データの閲覧権限とその管理者について見直したので、その検討過程と現時点での方針についての投稿になっています…

KAKEHASHI でテックリードをしている横田です。 KAKEHASHI に入社して早 5 年が経ちまして、色々な経緯から社内勉強会の運営をしてきました。 その中で感じた社内勉強会による共有知の有用性について、紹介させていただきたいと思います KAKEHASHI の勉強会について KAKEHASHI 社内では、エンジニア・非エンジニアに限らずさまざまな勉強会が立ち上がり、チームをまたがって学習する文化があります。 例えば、今まで私が参加した勉強会には以下のようなものがありました。 技術 Web API の設計 輪読会 ドメイン駆動設計 勉強会 実践的データ基盤への処方箋 輪読会 データ基盤勉…

こちらの記事は Databricks Advent Calendar 2022 の22日目の記事になります。 こんにちは、カケハシで Musubi Insight のバックエンドエンジニアをしている末松です。 カケハシでは 全社的なデータ活用基盤のプラットフォームとしてDatabricksを採用 しておりますが、それまでは Redash を利用していました。 Redash は Databricks社にM&A された背景もあり、基本的には Redash を使っていた感覚のまま Databricks SQL を使用することができます。 とはいえ細かなところに違いがあるので、カケハシ内でも Reda…

こちらの記事は、カケハシ Advent Calendar 2022の22日目の記事になります。 こんにちは。KAKEHASHIでおくすり連絡帳 Pocket Musubi というサービスを開発している牧野です。 この記事では、アジャイルに開発をする中で、リリースまでに数ヶ月を要するプロジェクトを進める場合に意識していることについて書かせていただきます。 小さく完了していく アジャイル開発を経験したことがあれば アジャイルソフトウェアの12の原則 を読んだことがある人は多いかと思います。 その中で 顧客満足を最優先し、価値のあるソフトウェアを早く継続的に提供します。 動くソフトウェアを、2-3週…

こちらの記事はカケハシ Advent Calendar 2022 の21日目の記事になります。 はじめに こんにちは、カケハシのデータ基盤チームのデータエンジニアの大木です。今年も残すところ10日ほどになりましが、皆さんいかがお過ごしでしょうか。 私はカケハシに入社したのが2022年の1月ですので、もうすぐ1年が経とうとしています。本当に月日の流れは早いものです。 個人的に入社1年という節目を迎えるということもあり、この記事では私がデータ基盤チームにJOINしてからの1年間のチーム活動を振り返ってみようと思います。 2022/01~ 入社、チームのミッション・コアバリューの設定 まずは私が入社…

こちらの記事は、 カケハシ Advent Calendar 2022 の20日目の記事になります。 はじめまして！ プラットフォームチームの筋肉大好き五十嵐です💪 この記事では自作キットのキーボードに入門して約4か月の私が、入門するきっかけや入門してどうだったのかお話します。 入門のきっかけ 分割型の自作キットのキーボードを導入した、と知人エンジニアが語っているのを聞いたのがきっかけでした。 それまで私はMacBook ProのUSキーボードを使っていましたが、ちょうどキーボード操作時の肘の置き場所に困っていた頃で、知人の話を聞いていく中で 分割型のキーボード導入したら、今使っている椅子のアー…

こちらの記事は カケハシ Advent Calendar 2022 の19日目の記事になります。 あっという間に2022年も終わりますね⛄️ プラットフォームチームの石黒です。 今年は遅ればせながらFF9をプレイしまして、トロフィーをゲットするためにフィールド上でモーグリのモグオをたてぶえで呼びつけ、「なんでもない」を繰り返して怒られてしまったときに、ふとLambdaのことを思い出しました。 AWS Lambdaは拡張性に優れたコンピューティングサービスですが、モグオと同じく呼び出し回数の制約があります（モグオには怒られるだけですが…）。 今回はLambdaの呼び出し回数にフォーカスして、スロ…

この記事は、カケハシ Advent Calendar 2022 の 18 日目 の記事になります。 はじめまして、こんにちは。 おくすり連絡帳「Pocket Musubi」というプロダクトで、エンジニアリングマネージャーをしています @hisasann と申します。 人にフォーカスした開発組織作りに力を入れ、楽しい技術集団を作り上げることに日々奮闘しております。 ぼくは、 20 年近く Web の業界でいて、ソフトウェアエンジニアとして開発をしてきました。 今でも第一線ではないですが、なるべくコードは書いていて、それは好きというのがそもそもですが、日々テクノロジーの変化を楽しんでいます。 特に…

こちらの記事は カケハシ Advent Calendar 2022 の17日目の記事になります。 こんにちは、カケハシで Musubi Insight のバックエンドエンジニアをしている末松です。 Musubi Insight に表示するデータは夜間の日次バッチで集計しているのですが、テスト・品質担保・パフォーマンスなどなど悩みが絶えません... 以前もバッチ処理のテストに関するブログを掲載しましたが、今回はパフォーマンスに関する記事になります！ https://kakehashi-dev.hatenablog.com/entry/2022/08/12/094856 Musubi Insigh…