Zenn
Open11

scala on databricks

Scala
#
spark
110416110416
  • Spark は「分散コンピューティングエンジン」
  • Databricks は 「マネージド Spark」

クラスタは管理が辛いのでその部分をDatabricks がよしなにやってくれる.

実体は AWS + Spark クラスタ + ストレージレイヤー + jupyter notebook + ml flow のようなもの.

アナリストとエンジニアのギャップを埋めて作成したモデルを運用に乗せやすくしてくれる.
gcp の kube や lightbend の akka serverless みたいに分散アーキテクチャの管理をよしなにやってくれるのでうれしい.

110416110416

Databricks はノートブックを job として実行できるので 開発者 と違って データサイエンティスト・アナリストはそれをそのままjobにしてしまいがち

これはいい面もあるが悪い面もある
良い面として、例えばセルごとのSparkタスクの実行結果やメトリクスが表示されるのでライトなチューニングがしやすい.

一方で、githubのインテグレーションは存在するがそれほどスムーズではない.

cli や rest API が提供されているのでプロダクション環境はGithubなどと連携してコードの管理をしたい.

マネージドによるイージーさと運用のシンプルさ、Databricks のサービスにロックインされてしまうことのバランスをうまくとりたい.

110416110416

job は notebook ベースか uber jar.
jar は s3 や gcs にアップロードして、そこから引っ張ってくる. 様々なストレージサービスとのインターフェースとなる dbfs(databricks file system?) と それにアクセスできる ライブラリ dbutils や databricks cli を使う.

Scala で executable な jarを sbt assembly で作ってup するとよさそう

jarをupする場合github actions などを使ってテスト・コードレビュー・CI/CDのフローをつくれば開発者フレンドリーになると思う

110416110416

ACID トランザクション、耐障害性を備えるストレージレイヤーに delta lake がある. これはイベントソーシングベースの永続化層. dbのstateとそれを変更する 一連の action によってデータを管理するので特定の時点のデータベースの状態を後から再現することが容易になる. これは git-like なインターフェースを備えたOSS のdb dolt のようで面白い. データベースがある意味イミュータブルなので監査の面でもメリットがあるかもしれない.

110416110416

Databricks 上では Scala 2.12 が主流の模様(´･ω･｀) エンタープライジーなかほりがする...

110416110416

一部の機能e.g. dbutils はdatabricksのruntime上に公開されているがライブラリとして積極的に公開されているわけではないので注意が必要。主要な機能はossだがところどころそうではないものもある。

110416110416

クラスタの管理しなくていいのはだいぶ楽だけど Spark が必要になるケースは意外と少ない.

110416110416

バッチ処理のスケジューラーやタスクグラフをGUI ぽちぽちで組めるのは便利.
データを集める処理でインフラ管理の無駄な手間が増えないのでデータサイエンティスト > データエンジニア な少数精鋭のチームでよさそう.

110416110416

Premium プランで SQL が使えるけど、既にある程度のデータ基盤があって開発者・分析者が Databricks を使うモチベが高くないなら SQL の機能を使えるメリットが薄そう...
Standard プランで、インフラ管理がいらない ETL とアドホックな分析用の マネージド Spark　として使う方が良さげ...

110416110416

テーブルの ZORDER BY で <col> を指定すると Z-Ordering will be ineffective, not collecting stats のエラーになるとき

まず、SQL でカラム <col> を先頭に持ってくる

ALTER TABLE <catalog>.<schema>.<table>
CHANGE COLUMN <col> <col> <col type>  FIRST
 --- 例: CHANGE COLUMN example_col example_col STRING FIRST


import com.databricks.sql.transaction.tahoe._
import org.apache.spark.sql.catalyst.TableIdentifier
import com.databricks.sql.transaction.tahoe.stats.StatisticsCollection

val tableName = "<catalog>.<schema>.<table>"

val deltaLog = DeltaLog.forTable(spark, TableIdentifier("<table>",Some("<schema>"),Some("<catalog>"))
// または val deltaLog = DeltaLog.forTable(spark, dataPath = "/mnt/<path/to/external/location>")

StatisticsCollection.recompute(spark,deltaLog)
110416110416

Read GCP BigQuery from Databricks on AWS

サービスアカウントに jobUser BigQuery Storage API やテーブルの閲覧権限が必要

Not found: Dataset xxx was not found in location US

  • Materialized view の読み込みに失敗しているケース(bigquery.table.create が不足していると表示される)
    • Spark から読み込む場合, Materialized view は一時テーブルに変換する必要がある. そのための bigquery.table.create 権限が不足している. デフォルトでは mat view が存在するプロジェクトに table を作成しようとする.
    • materializationProject, materializationDataset を指定することで指定したプロジェクト・データセットに mat view を一時テーブル化できる.
val df = spark
  .read
  .format("bigquery")
  .option("project", "<a gcp project where a materialized view exists>")
  .option("parentProject", "<the same or another gcp project where a service account for spark belongs to>")
  .option("viewsEnabled", true) // view 読み取りは明示的に設定しないといけない.
  .option("materializationProject","<same as parentProject>")
  .option("materializationDataset","<where a view is materialized>") // "<where a view is materialized>" に _sbc_0123456789XXXXX という名前の一時テーブルが作成される
  .option("table", table)
  .load()

References