2020-07-12

Apache BeamのDoFnで匿名クラス使う時はSerializableに気をつけようという話

BeamでDoFn書く時に、↓のように匿名クラスを使って書けます。

PCollection<Integer> wordLengths = words.apply(
  "ComputeWordLengths",                     // the transform name
  ParDo.of(new DoFn<String, Integer>() {    // a DoFn as an anonymous inner class instance
      @ProcessElement
      public void processElement(@Element String word, OutputReceiver<Integer> out) {
        out.output(word.length());
      }
    }));

しかし、Beam Programming Guideを良く見ると

Take care when declaring your function object inline by using an anonymous inner class instance. In a non-static context, your inner class instance will implicitly contain a pointer to the enclosing class and that class’ state. That enclosing class will also be serialized, and thus the same considerations that apply to the function object itself also apply to this outer class.

という注意書きがあります。

BeamのDoFnはSerializableである必要がある（ワーカーに送るので）
Javaの匿名クラスは、enclosing class（匿名クラスを含んでいるクラス）への、暗黙の参照を持つ（参考： SER05-J. 内部クラスのインスタンスをシリアライズしない）
ので、enclosing class（とそのフィールド）はSerializableである必要がある

というわけです。

気をつければ大丈夫な話ですが、

引っかかりやすい
ネストが深くなるので見にくい
匿名クラスの部分のテストしづらい

ので、DoFnを書く時、私は匿名クラスとしては書きたくない派です。

2020-07-12

Airflowの動的なDAGでclearする時の話

ハマったのでメモ。（Airflow 1.10.2/Cloud Composer）

状態

Airflow Variableにも基づき、Airflowのタスクを動的に作成していた
（Apache Airflow: Create dynamic DAG – Big Data & ETLのようにタスク外の部分でループしている感じ）
DAG Runが実行・修了
Airflow Variableを修正し、DAGにタスクインスタンスが追加される
（Airflow Variablesの変更によって追加されたTaskはstatusが空欄）

通常はclearでTask Instanceを再実行する事ができますが、今回の場合は対象のTask Instanceをclearしても実行されません。

f:id:toukoudo:20200712075431p:plain — Task Instanceが増えた状態のDAG Run

f:id:toukoudo:20200712075528p:plain — 増えたTask Instanceをclearした時のエラー

対応

対象のTask Instanceを「Mark Failed」にする
対象のTask Instanceをclearする
DAGがRunningになり、Task Instanceが実行される

増えたTask Instanceが複数ある時、どれか一つで1・2を実行すると、追加されたTask Instanceは全て実行されます。

原因

AirlfowはDAG Run・Task InstanceをDBのレコードとして持っている
clearを実行すると、Airflowは、Task Instanceのレコードを探しステータスを変える
DAGが変化した時、過去のDAG Runに対応するTask Instanceのレコードは追加されない（下図）
（画面からは追加されたように見えますが、見えているだけ）

ので、DAGが変化してTask Instanceが増えた時、増えたTask Instanceは単純にはClear出来ないです。

f:id:toukoudo:20200712081324p:plain — Task Instanceが増えた状態のtask_instanceテーブル

しかし、前述の対応を取ると、

Mark Failedすると、対象のTask Instanceのレコードが出来る
（=Clear出来るようになる）
ClearするとDAGがRunningになり、StatusがSuccessでもFailedでもないTask Instanceがスケジュール・実行される
（＝増えたTask Instanceが実行される）

となり、増えたTask Instanceが実行されます。

f:id:toukoudo:20200712081347p:plain — MarkFailed&Clear後のtask_instanceテーブル

2020-06-27

今日聞いたPodcast: Materialize (Software Engineering Daily)

softwareengineeringdaily.com

概要

テーマ：Streaming SQLエンジンのMaterialize
ゲスト：開発元の創業者Arjun Narayan（元Cockroach）と、Frank McSherry（元Microsoft）

materialize.io

Materializeの説明

What is Materialize? とかArchitecture OverviewとかWhat’s inside Materialize?がわかりやすいです。

ストリーミングデータ（Kafka）、RDBの両方のデータにSQLを投げられる
（DBのデータはDebeziumでChange Data Capture可）
普通のクエリも使えるが、materialized viewにすると継続的に更新し続けることができる（かつ早い）
PostgreSQL互換のSQLが使える（大変だったらしい）
Cloudサービスだけを提供

Streamingの話

Event timeとProcessing Timeの違い（Beamでよくでるやつ）
ラムダアーキテクチャの話
色々プラットフォームが出てフラグメンテーション化（Storm, Spark, Heron, Apex, Google Beam‥）

Materializeの歴史

Microsoftのnaiadプロジェクトが源流

論文
特許 US9165035B2 - Differential dataflow

その後Microsoftがプロジェクトを解散。しばらく後に、Materializeが生まれた。

その他・感想

スクリプト見ると30ページあってスゴイ。聞くだけで理解出来なかったけど、そんだけ長ければ‥
他のStreaming系（Apache Druid、Spark SQL、Beam SQL‥）との違い・使い分けが気になる
事例的な物ってあるのだろうか
人はなぜプロダクト名に一般名詞を付けるのか（検索しづらい）

2020-06-24

Google Cloud in Action読んだ

learning.oreilly.com

2018年に出たGoogle Cloud Platformの本です。
著者の John J. GeewaxさんはGoogleのエンジニアです
サンプルプログラムはNodeJS（あんまり依存しないと思いますが）です

以下のGCPサービスが紹介されており、

DB・ストレージ系(GCS.BigTable,Datastore,CloudSQL,Spanner)
Compute系（GCE,GKE,GAE,Cloud Function）
機械学習系（Cloud Vision, Cloud Natural Language, Cloud Speech, Cloud Translation, Cloud Machine Learning Engine）
ビッグデータ系（BigQuery, Pub/Sub, Dataflow）

それぞれのサービスに関して、

概要
特性（レイテンシ、スケール、耐久性）
合う合わない（小規模なTODOアプリ、大規模SNS、証券取引）
価格感

が説明されています。GCPのサービスの感じを掴むのにいい感じです。

個人的に勉強になったのは、Pub/Subの属性とメッセージデータの使い分けの記載です。

メッセージデータはデコード（Base64。任意で暗号化）が必要、属性は不要なので、メッセージのルーティング（※）的な事をするためのデータは属性に入れるのが、想定してる使い分けらしいです。

※サブスクライバーが処理すべきメッセージか判定したり、データ内容によって処理内容を変更する

2020-06-17

O'Reillyのサブスクリプション（Safari）登録してみた

すべての情報は2020/06時点のものです

Safariとは

www.oreilly.co.jp

O'Reillyが運営している、オンライン学習プラットフォームです。

何が見れんの

O'Reillyの本
O'Reilly以外の本
カンファレンスのビデオ（Strataとか）
インタラクティブな学習ツール（Katacoda）

などが使えます。

O'reilly以外の本は、例えば、Efffective Java（ピアソン）、Scrum Book（Pragmatic Bookshelf）などが読めます
基本英語の情報のみで、O'Reillyの本であっても、水野さんのWeb API: The Good Partsや、渋川さんのReal World HTTPは読めません

(2020/6/17時点のコンテンツの状況。英語だと５万タイトル)

f:id:toukoudo:20200617211638p:plain

でもお高いんでしょ

www.oreilly.com

O'reilly直接では、月$49もしくは年$499です。

月1・2冊O'reillyの本読めば元とれるレベルです。

www.acm.org

ACM（Professinal Membership）に加入すると、Safariを使うことが出来ます。

年$99なので、年2・3冊読めば元取れるレベルです。

なお、ACMに入るには会員資格があります（ただし審査はない？）：

Membership Qualifications
You must satisfy one of the qualifications below:

Bachelor's Degree (in any subject area); or

Equivalent Level of Education; or

Two years full-time employment in the IT field.