Serverless Framework+mangum+FastAPIで、より快適なPython API開発環境を作る

はじめに

最近ハイボールにハマっているSREのたっち(@TatchNicolas)です。

昨日オンライン開催されたJAWS DAYS 2020にて、JX通信社もサーバレスをテーマとして発表をしました。(by 植本さん

発表でもありましたように、上記プロジェクトにおいて開発当時はスピードを優先してプロジェクトメンバーの手に馴染んでいて分担もしやすいフレームワークとしてFlaskを採用しました。

一方で、JX通信社としてはFlaskよりもFastAPIを使うプロジェクトが増えてきており、今後もその傾向は続く見込みです。

そこで、特設ページ作成やAPI提供など初動としての開発が一段落したのを機に、JAWS DAYSで発表した仕組みを今後のために発展させる検証をしたので紹介します。

TL; DR;

  • JAWSでは Serverless Framework+awsgi+Flaskな構成でスピーディにコロナ特設ページ向けAPIを作った話をした
    • 緊急性の高いプロジェクトであったが、非常にスピード感のある開発ができた
  • Flaskに代わるフレームワークとしてJX通信社ではFastAPIが流行中
    • OpenAPIドキュメント自動生成、Request/Responseをクラスとして定義するなどカッチリと開発ができる
    • しかしFlaskライクな軽量フレームワークで、シンプルで書きやすい
  • そこで、FastAPIを使うパターンでも前述の良い開発者体験を提供する仕組みを作った

サンプルコードはこちらにおいてあります。

github.com

前提

awsgiはAPIGatewayからLambda Proxy Integrationに渡されるイベントをWSGI*1アプリケーションが理解できる形式に変換し、またWSGIアプリケーションが返すレスポンスをLambda Proxy Integrationに従った形式に変換してくれる便利なツールです。

github.com

WSGIの精神的後継として*2、asyncに対応したASGIが登場しました。そのASGIに対応したフレームワーク(FastAPIなど)とLambda Proxy Integrationの間でアダプタとして動いてくれるのが、mangumです。

github.com

ざくっとまとめると以下のようになります。

WSGI ASGI
フレームワーク例 Flask, Django*3 FastAPI,Sanic,Responder
Lambda Proxyアダプタ awsgi mangum
(HTTP Server)*4*5 gunicorn, uWSGIほか uvicorn

それでは早速やってみましょう。

やってみる

今回用意したサンプルコードはdocker-composeで動くようにしてあるので、下記のリポジトリをクローンしたら下記の要領で起動してみて下さい。

$ docker-compose up -d
$ docker-compsoe exam_results python -c 'from main import init_ddb_local; init_ddb_local()'

実際のコードを見ていきましょう。

# exam_results/main.py

(省略)

from fastapi import FastAPI, HTTPException
from mangum import Mangum


app = FastAPI()  # FastAPIのインスタンス

(省略)

handler = Mangum(app, False)  # FastAPIのインスタンスをMangumのコンストラクタに渡して、handlerとして外から読めるようにしておく

(省略)

exam_results/main.py のに app という変数名でASGI準拠のアプリケーション(=FastAPIのインスタンス)を作り、それを引数として作ったMangumのインスタンスを handler として定義します。

これを serverless.yml の中でhandlerとして指定することで、APIGatewayから発生したイベントをMangumが受け取り、FastAPIに渡してくれるというわけです。

# serverless.yml
(省略)
functions:
  exam_results:
    events:
    - http:
        path: /{path+}
        method: GET
        private: false
        cors: true
    handler: exam_results.main.handler  # ←ココ
    environment:
      PYTHONPATH: exam_results
      DDB_TABLE: ${self:custom.envMapping.${self:provider.stage}.DDB_TABLE}
    role: ManageStudentsRole
(省略)

細かなimportや環境変数設定は実際のサンプルコードを参照してください。

起動時の二行目のコマンドで、下記の関数を呼び出しています。

def init_ddb_local():
    if ExamResultsTable.Meta.host and not ExamResultsTable.exists():
        print('creating a table...')
        ExamResultsTable.create_table(
            read_capacity_units=1,
            write_capacity_units=1,
            wait=True
        )
        print('Done.')

最初のif文でテーブルのモデル(ExamResultsTable)の内部クラスMetaのなかで、「環境変数から DDB_HOST が指定されている」かつ「テーブルがまだ存在しない」場合のみPynamoDBを使って create_table しています。

実際にリクエストをしてみましょう。

$ curl -X POST -H "Content-Type: application/json" -d '{"name":"Alice","subject":"math","score":50}' localhost:8001/scores
{"name":"Alice","subject":"math","score":50}
$ curl -X POST -H "Content-Type: application/json" -d '{"name":"Bob","subject":"history","score":49}' localhost:8001/scores
{"name":"Bob","subject":"history","score":49}

$ curl localhost:8001/scores?student=Alice
{"exam_results":[{"name":"Alice","subject":"math","score":50}]}
$ curl localhost:8001/scores?subject=history
{"exam_results":[{"name":"Bob","subject":"history","score":49}]}

次に、このアプリケーションをAWSにデプロイしてみます。(sls コマンドのインストールやAWS操作のための権限は先に済ませておいてください。)

$ sls deploy --stage=dev
Serverless: Generating requirements.txt from pyproject.toml...
(いっぱいメッセージが出る)
Service Information
service: fantastic-service-with-fastapi
stage: dev
region: ap-northeast-1
stack: fantastic-service-with-fastapi-dev
resources: 13
api keys:
  None
endpoints:
  GET - https://xxxxx.execute-api.ap-northeast-1.amazonaws.com/dev/{path+}
  POST - https://xxxxx.execute-api.ap-northeast-1.amazonaws.com/dev/{path+}
functions:
  exam_results: fantastic-service-with-fastapi-dev-exam_results
layers:
  None
Serverless: Removing old service artifacts from S3...
Serverless: Run the "serverless" command to setup monitoring, troubleshooting and testing.

endpoints にあるURLに対して、localhost:8001 に対して行ったのと同じ操作ができると思います。

あとは sls deploy --stage=devdevstg prd に書き換えて実行すれば、それぞれ独立した環境を作成することができます。 コマンドがシンプルなのでCI/CDも簡単に書くことができるでしょう。

また、/docs にアクセスするとOpenAPIのドキュメントが自動生成されています。

f:id:TatchNicolas:20200323212131p:plain
自動生成されたdocs

何が嬉しいのか

開発スピードを高く保てる

デプロイも早く上手に使えばコストの運用負担も大幅に軽減されるLambdaと、ローカルでもガンガン開発しやすいWebアプリケーションフレームワークのいいとこ取りをすることができます。また、FastAPIの提供する「Web APIを作る」ために便利な機能を活用することができます。

Lambdaを手元で動かしたいという点で、本記事と似たことを実現しようとしているSAM Localは「Lambdaっぽいなにかを手元で動かす」発想です。一方で今回のやり方は「ローカルで動いているアプリケーションをASGI(FlaskであればWSGI)をインターフェイスとしてLambdaに載せる」という方法です。そのため、serverless.yaml 内ではhttpイベントのパスを{path+} 、 メソッドをANYにしておくことでL7の仕事をAPIGatewayではなくLambda内のアプリケーションに任せています。

awsgiもmangumも(gunicornやuvicornといった)HTTP Serverを起動するわけではなく、単にイベントを変換しているだけなのでオーバヘッドもほとんどゼロなため、Lambdaがキチンとスケールしてくれればパフォーマンスの心配はありません。

いつでも実行環境を載せ替えられる

たとえば運用しつつ何らかの事情で「Lambdaじゃキツイな」となってきたらECSやk8sなどコンテナベースの実行環境に載せかえることも容易にできます。その際にアプリケーションは変更する必要がありません。ローカルで動かしていたDockerイメージをそのまま、またはDistrolessなどにちょっと書き換えるだけで引っ越すことができます。

今回は最小限の構成にするために、Pythonのコードはすべてexam_results/main.pyに詰め込みました。しかし実際のアプリケーションではもっと真面目にファイル・ディレクトリの構成を切り分けると思います。その場合はLambda用(exam_results/lambda.py)とDocker用(exam_results/docker.py)に入口をそもそも分けてしまったり、Poetryのextrasを使って依存ライブラリ管理を分けたりしても良いでしょう。

ハマったところ/改善したいところ

serverless-python-requirementsのモジュールとPoetryがうまく連携しない

パッケージ管理およびvenvの管理にはPoetryを使っていたのですが、Serverless Framework側でモジュール機能を使うと、プラグインが期待通りに動作せず*6にpackageに依存ライブラリを含めてくれません。

PoetryだけではなくPipenvでも同じ問題があり、こちらはIssue報告されています。(執筆時点で未解決)

github.com

Pipfileやpyproject.tomlでなくrequirements.txtであれば individually: true のオプションがちゃんと効いてLambda関数単位で依存関係を解決してくれるようです。なので、CI/CDのタイミングでコマンドを一行足してモジュールごとにPoetryにrequirements.txtを生成させば解決します。

ただし今回のサンプルコードでは、1スタック1関数として作ることで回避しました。*7

DynamoDB LocalとPynamoDBでスキーマ定義が二重管理になってしまう

宣言的に書くためにDynamoDBのスキーマ定義はserverless.ymlに書きたいところですが、それをローカル開発時のDynamoDBにそのまま適用することはできません。

今回アプリケーションはPynamoDBを使ってDynamoDBとやりとりをしており、PynamoDBのモデルはserverless.ymlで定義したスキーマと一致するように書いています。そこで、簡単に初期化関数を定義してPynamodDBの機能を使ってローカルのテーブルを作りました。

しかしこれは実際のアプリケーションには必要ないある種「余計な」コードがソースの中に紛れ込んでしまっています。serverless.ymlに書いた定義をそのままローカル開発環境にも適用できればよいのですが、スマートな方法があれば改善したいです。*8

さいごに

サーバレスはその恩恵と引き換えに、どうしてもプラットフォームの制約を受け入れる必要が生じます。そんな制約の一つが「イベントという形で入力を渡されるので、Lambdaとしての書き方を強制されて慣れ親しんだフレームワークとは勝手が異なる」ことだと思います。

しかしWSGI/ASGIというインターフェイス(とそれらを繋いでくれるmangumのようなツール)を活用することで、ローカルで作業のしやすいコンテナベースの開発とサーバレスな開発をシームレスに切り替えられます。JX通信社のプロダクトではECS+RDSで一部補助的にLambdaを使う構成が定番でしたが、最近はサーバレスを積極的に活用する場面も増えてきました。

今回はそんな社内の技術スタックの移り変わりにあわせた開発ができるように工夫してみたことの記録でした。少しでも皆さんの参考になれば幸いです。

*1:Pythonの標準仕様として用意されているWebアプリケーションのインターフェース定義で、Ruby屋さんにとってのRackにあたります

*2:公式Docより: ASGI (Asynchronous Server Gateway Interface) is a spiritual successor to WSGI, intended to provide a standard interface between async-capable Python web servers, frameworks, and applications.

*3:Django 3.0からASGIに対応していく方針ですが、現時点で限定的です。詳しくは https://docs.djangoproject.com/ja/3.0/releases/3.0/#asgi-support

*4:Ruby屋さんにとってのunicorn

*5:今回はHTTP Serverは使わずに素のWSGI/ASGIアプリケーションとして動かして、mangumにブリッジしてもらいます

*6:プラグイン側のコードをみたところservicePathにプロジェクトルートが渡されてしまいpyproject.tomlを見つけられないことが原因のようでした。lib/pip.jsではmodulePathという変数にモジュールのディレクトリ名が入ってくるのでうまく連結できれば動かせたかもしれません

*7:厳密にマイクロサービスするなら永続化層も共有すべきではないという考え方もあるので、1つのスタックに1つのLambda関数を良しとしました

*8:たとえばserverless.ymlをパースして、テーブルの有無をチェックして必要に応じて作成するなどできると思います

Goの並行処理について

はじめに

こんにちは。サーバーサイドエンジニアインターンでお世話になっている杉山と申します。 今回はオライリー・ジャパン社より出版されている『Go言語による並行処理』を読み勉強したことについて書いていきたいと思います。

Goにおける並行処理

GoはCPUのコア数が複数だった場合並列処理になります。 もし1コアのマシンで並列処理を行う場合それは並列処理ではなく素早く順に実行しているだけのようです。

並列処理と並行処理の違い

並列処理と並行処理の違いについては『Go言語による並行処理』では 『並列性はコードの性質を指し、並列性は動作しているプログラムの性質を指します』 とありますが分かりづらいので調べてみると

  • 並列処理 複数の命令の流れを同時に実行すること

  • 並行処理 コンピュータの単一の処理装置を複数の命令の流れで共有し、同時に実行状態に置くこと

このような違いがあるみたいです。イメージで説明すると以下のようになります

f:id:sugi1208:20200323174721p:plain

並列処理で気をつけること

デッドロック

  • プロセスなどの処理が互いの処理終了を待ち、どの処理も先に進めなくなってしまうこと

ゴールーチンで同じ変数へのデータ競合

  • 解決方法
    1つの変数には1つのゴールーチンからアクセスする
    ロックをとる
    チャネルを使う

チャネル型とは

チャネルはゴールーチン間でのメッセージを共有するためのもの

  • 特徴
    送受信時にブロックできる
    送信時にチャネルのバッファが一杯だとブロックする
    受信時にチャネル内が空だとブロックする

並列処理を使って大量のデータを書き込んでみる

今回は東京の気温データ(8785件)をMySqlに書き込んでいます。 また今回のコードに関しましては並列処理の例ということでfor文の中にてinsertしております。ご了承ください。

試しに自分で書いてみるコードがこちらです。

package main

import (
    "encoding/json"
    "fmt"
    "io/ioutil"
    "log"
    "os"
    "sync"

    "github.com/jinzhu/gorm"
    _ "github.com/jinzhu/gorm/dialects/mysql"
)

type Weather struct {
    Time    string `json:"time"`
    Temperature string `json:"temperature"`
}

func main() {

    client, _ := gorm.Open("mysql", "root:root@tcp([mysql]:3306)/weather?charset=utf8mb4&parseTime=true")

    client.AutoMigrate(&Weather{})

    ch := make(chan int, 4)
    wg := sync.WaitGroup{}

    account_json, err := ioutil.ReadFile("data.json")
    if err != nil {
        fmt.Println(err.Error())
    }
    var data []Weather
    if err := json.Unmarshal(account_json, &data); err != nil {
        log.Fatal(err)
    }
    for _, name := range data {
        ch <- 1
        wg.Add(1)
        go func(name Weather, client *gorm.DB) {
            defer wg.Done()
            client.Create(&name)
            <-ch
        }(name, client)
    }
    wg.Wait()
    client.Close()

}

変数dataに気象データを入れてfor文を使用し書き込みの処理をしています。

上記のコードの良くないと思う点

  • ただforで回して書き込むより少し早くしただけのコードになって安定したコードではない

  • 上記のコードには該当しないがデッドロックが起きた場合を考慮していない

  • insert処理が途中で止まってしまった場合処理が終了しない

上記の点を改善していきます

本を読んで書いてあったことを試してみる

次に本で読んでみて学んだ事を踏まえてコードを書いてみました。
今回はハートビートと言われる並列処理で用いられる手法を使用します。

ハートビートとは

ハートビートとはその名の通り人間の心拍のような物です。並列処理内部で外部にそのプロセスが生きているかを伝える方法です。

ハートビート使用した場合のメリット

今回の場合ハートビートは一定時間毎に外部にプロセスが生きているか伝えています。 ハートビートを使用することによって並列処理内部にて何しらの処理が止まってしまい外部に一定時間以上外部にプロセスが生きているか伝えられていない場合や、デッドロックしてしまった場合に処理を終了させることなどができます。

今回書いた全体のコードはこちらになります

こちらでは一部抜粋して説明していきます。

func main(){

    ----------省略----------

    done := make(chan interface{})

    defer close(done)

    hertbeat, results, errch := insert(done, data, client)

    <-hertbeat   //最初のハートビートがゴルーチンに入ったことを受け取ります

    i := 0

    for {  // ここはhertbeatの受け取りなどの処理
        select {
        case r, ok := <-results: //ここでは書き込んだ内容と書き込むはずの内容の比較をしています
            if ok == false {
                return
            } else if Weather := data[i]; Weather != r {
                log.Fatal("MismatchingError", Weather)
            }
            i++
        case <-errch:  
            log.Fatal(errch)
        case <-hertbeat:  //一定時間ごとにハートビートを受け取ります。
        case <-time.After(5 * time.Second):    //ここでは上記の処理が一定時間行われない場合にタイムアウトするようにいています。
            log.Fatal("Timeout")
        }
    }

}

func insert(done <-chan interface{}, data []Weather, client *gorm.DB) (<-chan interface{}, <-chan Weather, <-chan error) {
    hertbeat := make(chan interface{}, 1) 
    weatherch := make(chan Weather)
    errch := make(chan error)

    go func() {
        defer close(hertbeat)
        defer close(weatherch)

        beat := time.Tick(time.Second)

    Loop:
        for _, name := range data {
            for {

                select {
                case <-done:
                    return
                case <-beat:
                    select {
                    case hertbeat <- struct{}{}:  //ハートビートを一定時間ごとに送っています  
                default:
                    }
                case weatherch <- name:
                    client := client.Create(&name) //ここではDBに対する書き込みです
                    if client.Error != nil {
                        errch <- client.Error // ここではエラーハンドリングをしています。そのまま表示させてもよかったのですが今回はこのような形で書きました
                    }

                    continue Loop
                }
            }
        }
    }()

    return hertbeat, weatherch, errch    //   hertbeat, weatherch, errchのchanを返しています
}

メインの処理はinsert関数の中ではhertbeat等のチャネルを作成しinsertの処理をgo func() を使用し、main関数の下のforにてheatbeatを受け取る等の処理と並列に実行しています。

またコメントにもありますがheatbeatが一定時間受け取れていない場合selectの最後にタイムアウトするような処理を書くことによって処理を中断させることができます。

並列処理内部にて処理の停止、もしくはデッドロックが起きた場合の処理の比較

ここでは自分が書いたコードとハートビートを使用したコードが並列処理内部にて処理が止まってしまった場合にとのような動きをするかを比較していきます。

ここではデットロックや内部の処理の停止の再現としてtime.Sleep()を使用し60秒程の時間、処理を停止させます。

自分の書いたコードの場合

func main(){

    ----------省略----------
    for _, name := range data {
        ch <- 1
        wg.Add(1)
        go func(name Hoge, client *gorm.DB) {
            defer wg.Done()
            time.Sleep(60 * time.Second)//  この部分を追加し、処理を60秒停止させています
            client.Create(&name)
            <-ch
        }(name, client)
    }
    
    ----------省略----------

今回は60秒停止ということにしてありますがもし完全に停止してしまった場合(デッドロック等)、処理は終わらずにタイムアウトすることもありません。

ハートビートを使用したコードの場合

func main(){    
    ----------省略----------
    case weatherch <- name:
    time.Sleep(60 * time.Second)//  この部分追加し、処理を60秒停止させています
    client := client.Create(&name)
    if client.Error != nil {
        errch <- client.Error  
    }
  ----------省略----------

ハートビートが5秒間外部に送信されない場合以下のようにタイムアウトさせることができます。

2020/03/23 07:57:20 Timeout
exit status 1

このように並列処理内部にて何かしらのトラブルがあり、処理が停止した場合ハートビートを使用することによって処理を中断することができます。

最後に

Goの並列処理は少ししか触れていなかったので今回勉強して並行処理について深く知れました。『Go言語による並行処理』ではGoに限らず並行処理自体の勉強にもなるので難しいですがとても勉強になる本でした。まだ理解が浅い部分もあるので繰り返し読み勉強していきたいと思います。またこの場をお借りして勉強の機会を下さった会社の方々にお礼を申し上げたいと思います。ありがとうございました。

JX通信社の取り組みとメンバーの優しさがデブサミの登壇を支えてくれたという話

先月に引き続きデブサミの話で失礼します.

JX通信社でシニア・エンジニアをしています, @shinyorke(しんよーく)と申します. 最近引っ越した新オフィスの近くにあるビャンビャン麺がマイブームです.

前回のエントリーでもご紹介させてもらいましたが, 先日開催されました「Developers Summit 2020」にて, 自分のキャリアを基にした未来の話をさせていただきました.

f:id:shinyorke:20200218111119j:plain
満員御礼でした(感謝)

満員御礼いや...立ち見の方も出たぐらいに大盛況で感謝しております.*1

来ていただいた皆さま誠にありがとうございました!

どういった発表をして, どんな準備をしていたのか?についてはスライドブログをご覧いただくとして.*2

こちらのエントリーでは,

デブサミ(と後日登壇のスポアナ)の発表は「JX通信社」の取り組みとサポート無しではできなかったんですよ!

という大事な話を,

  • JX通信社のメンバーが活用できる制度
  • エンジニアチーム独自の取り組み
  • 優しさあふれる周りのサポート

といった制度・文化の側面でご紹介いたします.

なお, 今回はTechな話題がほぼないエントリーとなります :bow:

TL;DR

  • イベント・勉強会参加は業務時間扱いなので気楽に参加できますよ!
  • さらに土日・祝日に登壇をした場合, しっかり代休もらえます
  • プロジェクト管理(Jira)を使わせてもらったり, 発表練習・Slackなどを通じて会社・メンバーから優しいフォローいただきました(感謝)

おしながき

勉強会・イベントに関するサポートについて

JX通信社では全社的な制度として,

勤務時間中の勉強会等への参加自由・参加費補助

知識・スキルの向上のため、勤務時間中に社外で催される勉強会等に参加したい場合は、自由に参加することができます。また、勉強会の参加費は全額を経費として処理できます(会社負担)。

※引用元: JX通信社HP「採用情報」

があります.

デブサミは平日の日中に行われるイベントではありますが, 私はこちらの制度のおかげで登壇した2日目はフル参加, 初日は午後半日参加できました.*3

複数人が応援に来てくれた(感謝)

この制度を活用して自分が登壇した当日は4名のメンバーが応援に駆けつけてくれました.

f:id:shinyorke:20200218100520j:plain
撮影とか色々と手伝ってもらいました.

こんな感じで沢山写真を撮ってもらったり,

  • 他の発表を聞いたり, ブースを回った感想を共有してもらったり
  • 様子をSlackに呟いて社内のメンバーに共有したり*4

などなど, 私一人ではとても手が回らない所を助けてもらったりしました.

土日・祝日に登壇すると代休を取れる

さらに最近,

土日・祝日に登壇を行うとイベント時間に応じて代休取得可能

という制度もできました(条件を満たした場合に限る)*5.

実はデブサミに登壇した2/14から中一日で「Sports Analyst Meetup(スポアナ)」というイベントでロングトークを2/16にさせていただいたのですが, こちらのトークが日曜日にあったこと, お休みの条件も満たしていたため翌日はしっかり休んで疲れを取ることができました.

準備段階でのサポート

また, デブサミの登壇前は以下のサポートをいただきました.

  • プロジェクト管理ツール「Jira」の利用許可
  • 社内勉強会やSlackチャンネルを通じたサポート

Jiraでデブサミのタスクを管理

JX通信社ではJiraをプロジェクト管理ツールとして使っています.

「デブサミの管理でも使いたいなあ」と相談した所, 二つ返事でOKもらったので準備段階からフル活用させてもらいました.

f:id:shinyorke:20200218103135p:plain
スプリント単位でタスクを回していました

Jiraにタスク管理を集中できたので本番まで大きなトラブルもなくできました.

また, こうしてメトリクスやレポートが残っていくのでまだJiraを活用しきれていないチームやメンバーに向けてのサンプルとして活用もできそうとも思いました.

利用させてもらったぶん, 知見をこっちから発信していくぞ!

月次勉強会・Slackを通じたメンバーのサポート

これは前回のエントリーでもちょっと紹介させてもらった話でもあります.

tech.jxpress.net

デブサミ本番で披露したデモアプリケーションは, 月次の社内勉強会で共有したものをほぼそのまま披露できました.

開発・企画段階でSlackに相談やネタを披露すると光の速さでレスが来たり, コードレビューするよ!と手を上げてくれたりとノウハウ的・心理的にも色々と助けてもらいました.

また, 月次の勉強会でのフィードバックも多くこれらがあったことにより本番も安心して望むことができました.

もっとアウトプットを出していくぞ

というわけでこのエントリーでは「JX通信社がアウトプットを出す際のサポートや制度」といった文脈を紹介させてもらいました.

これらの制度を活用し, もっとアウトプットを出すことによって成長に繋がるような流れが最近できつつあるので今後のJX通信社およびこのブログにもご期待ください.

最後までお付き合いいただきありがとうございました.

*1:正式な入場者数の数字は把握していませんが, 満席率100%超えだったのは確かです.

*2:なので発表内容そのものやそれに付随するエピソードについては割愛いたします. リンク先のスライドもしくはブログをご覧ください(手前味噌).

*3:初日が半日だったのは, 自分の作業の都合で自ら選んだもので, やろうと思えばフル参加できました

*4:その他, 一部のメンバーはツイッターでも呟いてもらって盛り上げに貢献してもらいました(圧倒的感謝)

*5:具体的には会社およびサービスの認知に繋がるような紹介が必要となります

Nuxt.js + FastAPIを使ったデータエンジニアリングなデモ作り - 社内勉強会でデブサミのデモをしました

(今更ですが)新年あけましておめでとうございます!

JX通信社でシニア・エンジニアをしています, @shinyorke(しんよーく)と申します.

最近は週に2, 3回, ジムで10kmちょい走っています.*1

JX通信社のエンジニアチームでは, 月に一度みんなが集まる月次勉強会というイベントがあります(基本的に第2金曜日開催)*2.

tech.jxpress.net

※過去の開催レポです

2020年初(かつ, 飯田橋オフィス最後*3)の勉強会は,

「普及したいことや年末年始に勉強したことなどを発表するLT大会」

ということで, 私は

  • デブサミ2020登壇時に披露するデモアプリを披露
  • 弊社プロダクトでも使っているFastAPI僕もやりました&Nuxt Core UI ええやで!っていう布教
  • (ちょっとだけ)野球選手の評価指標を紹介

という発表をさせてもらいました.

このエントリーではそんな発表内容のサマリーおよび, 勉強会の様子をちょびっと紹介したいと思います.

Nuxt.js + FastAPIを使ったデータエンジニアリングなデモ + 月次勉強会レポート

というわけで自分の発表内容を(デブサミ本番に支障をきたさない程度に)紹介します.

なぜデモを作ったのか

いきなり宣伝で恐縮ですが, 「Developers Summit 2020」の2日目, 2/14にこちらのテーマでお話させてもらうことになりました.

event.shoeisha.jp

カテゴリーが「エンジニアの生き方」で, 話す内容も細かい技術内容よりも私自身のキャリアの話なので一見するとデモなんていらないのでは?という感じなのですが,

  • 私のキャリアの話にうまくセイバーメトリクス*4の話を絡めてデータサイエンス&エンジニアリングな味付けが欲しい
  • 自分はエンジニアであり, デブサミは名前の通り「開発者の会議・お祭り」なので開発者らしい振る舞いをしたい
  • それより何より, 年末年始という時間もあったしガッツリとゼロベースで(小さくても)プロダクト作りたい!

という自分の趣味モチベーションでエイヤッと作って当日披露するまでに至りました.

デモの技術選定&構成

デモの技術選定ですが,

  • デモはローカル(Docker Machine)上でやるにしても, AWSなりGCPなりにローンチすることに備えてサーバレスでイケるような構成
  • 可能な限り, 「自分がまだ未経験」もしくは「チームメンバーにオススメしたい」Framework・ライブラリを使う
  • 「Done is better than perfect.(完璧を目指すよりまず終わらせろ)」少なくとも最初のデモは要件をきっちり実装&見せられるような質とスピードが出せるものを重視

ということで, 以下のような構成で作りました.

f:id:shinyorke:20200122202113p:plain

マイクロサービスというほどの物ではありませんが, Containerベースでフロント・バックエンド・分析の各レイヤーで必要以上に相互依存をしないことを目指しやりきりました.

Nuxt Core UIによる管理画面テンプレ(からのデータ表示アプリ開発)

JX通信社では, iViewと呼ばれる管理画面フレームワークを使って管理画面を作ることが多いです.

tech.jxpress.net

が, 「選択肢としてこんなのもあるよー」ということで以前私がPyCon JP 2018で紹介させてもらったNuxt Core UI という, Nuxt.jsベースの管理画面フレームワークを紹介ということでこちらを使いました.*5

shinyorke.hatenablog.com

今回は以前作ったものをベースに,

  • Nuxt.jsのバージョンをUpgrade
  • 状態保持・保存をちゃんとVuexで実装
  • 作り的にイケてなかった細かい所をブラッシュアップ

という感じで約1年半ぶりに更新して望みました.

発表したときもグラフが動いた時にオッていうリアクションが小さいながらもあったりよくやったなって思いました(自画自賛).

FastAPIによるザクッとAPI構築

バックエンドのAPIはホントはGolangで作る予定でしたが,

  • Golang自体は初めてではないが3, 4年近いブランクがあり「Done is better than perfect.」が守れるかあやしい
  • 作ってる途中で仕様がゴリゴリ変わる可能性が高いデモなので一旦は慣れてるPythonで書いたほうが効率は良さそう
  • Python版は「動くアプリそのものだけど設計図」扱いにしてどっかのタイミングで丸っとGolangに!

ってことで急遽Pythonによる開発にシフトしました.

最初はresponderを使うつもりでしたが, 年末年始に有志メンバーでもくもく会していた時に,

今, 社内ではFastAPIアツいですよ!

と言われたので試しにFastAPIを使ってみることにしました.

これは選択肢として大正解で, 紹介してもらったその日の内にAPIの実装が8割ほど終わりました.

shinyorke.hatenablog.com

機能拡張もブラッシュアップもいい感じになった&目論見も達成し, Nuxt Core UIによるフロントエンド開発に終始できたのはホント良かったです.

類似性スコアで似ている選手のリコメンド

実はキモな機能である「似ている選手リコメンド」については「類似性スコア」というスポーツ界隈で使われているモデルを使いました.

shinyorke.hatenablog.com

上記のエントリー作ったものを元に,

  • データを2019シーズンのモノに差し替え
  • 約19,000人の選手のスコア計算のためDocker化してGCE + GCS上でゴリッと計算

しました.

年をまたぐぐらいのタイミングで半日ほどかけてやりました.

当日の成果

というデモを踏まえて話した内容がこちらになります.

約5分の発表と同じくらいの質疑応答がありました.

  • 野球の指標や類似性スコアに対する質問
  • そもそもデータはどこにある*6
  • Web系の他のリコメンドとかにも似ていて面白い

などなど色々フィードバックがあってよかったです.

おかげさまでデモの披露も最初の素振りもいい感じでした.

その他の発表&次回に向けて

なお, この会はLT会でいろんなメンバーのいろんな発表がありました.

  • Amplifyを使ったアプリ開発の勘どころ
  • 既存プロダクトを使いたい技術でゼロベースで作り直す
  • 認定スクラムマスター研修レポート
  • 過去にいたチームのアーキテクチャ話
  • GISデータの扱い方
  • エンジニアによく効く作文技術のはなし

全てが面白く, 中にはお腹を抱えて笑える(もちろん学びも深い)モノもあり,すべてを紹介したいのですが大人の事情で字数が足りないのでやめておきます苦笑

全体としては, それぞれの技術的な議論や興味関心で盛り上がり和気あいあいとした会でした, 個人的には2, 3ヶ月に一度こういうLT会があってもいいかなと...

また, 今回は社員のみの発表でしたがインターン生の成果やドヤる姿もみたいな!って思ったのでこちらもいい感じにジョインできるような感じにしていきたいですね.

さいごに - デブサミで会いましょう!

ということで簡単ではありましたが, 月次勉強会のレポを通じてJX通信社のエンジニア文化を紹介させてもらいました.

中途採用およびインターンも募集していますので, 今回のレポでご興味を持った方はぜひカジュアル面談等で来てもらえると嬉しいです.

jobs.jxpress.net

また私個人としては, 「Developers Summit 2020」で登場するのでこちらもよろしくおねがいします(大切なので二度言う)

最後までお読みいただきありがとうございました&次のエントリーは(来週から稼働する)新オフィスからお届けいたします!

*1:ちなみに走ってる理由は, 個人OKRのKR3にあります.

*2:ちなみに私は入社直後にPySparkを交えたデータ基盤入門みたいな話をしました&その成果の一部はアドベントカレンダーでも紹介させてもらいました(元々は勉強会ネタだったのです).

*3:今月中(というより今週末)に新オフィスにお引越しします

*4:書籍・映画「マネーボール」で有名になった野球の統計学的分析の概念と手法で昨今は「ピープル・アナリティクス」という人事版データサイエンスの元祖・教科書になっている分野でもあります.

*5:本当はiViewでゼロから作りたいお気持ちもありましたが時間足りないなってことで断腸の思いで断念orz

*6:念の為断っておくと今回の分析はメジャーリーグのオープンデータを使っています.

闇の Slack 魔術に対抗する Python 防衛術

この記事は、Slack Advent CalendarJX 通信社 Advent Calendarの最終日です。

メリークリスマス! 素敵なクリスマスをお過ごしでしょうか。取締役の小笠原(@yamitzky)です。

突然ですが、みなさん、ダークモードは好きですか? ダークモードは昨今のソフトウェアのトレンドで、Slack のデスクトップ版も今年の 9 月にダークモードに対応しました。

slackhq.com

しかし Slack をダークモードに設定すると、透過背景・黒文字なカスタム絵文字が見づらいという問題がありました。Slack が仕事のワークフローの中心にある JX 通信社にとって、これは死活問題です。

f:id:yamitzky:20191225003235p:plain

そこで、Slack の闇の魔術(ダークモード)に Python で対抗し、これらの絵文字が見えるようにしたいと思います!

方針

基本方針は簡単です。黒背景に黒文字だと見づらいのが原因なので、背景を白でベタ塗りした、ダークモードフレンドリーな絵文字を作ります。

f:id:yamitzky:20191225003312p:plain

この問題を分解し、次のような処理を Python で行うことを考えます。

  • Slack の API を使い、絵文字の一覧を取得
  • 取得した絵文字のうち、透過背景・黒文字なものを抽出
  • 白背景を合成した絵文字を生成

本当であれば、Python 経由で絵文字を更新したいところですが、公式の API ドキュメントに記載はなく、xoxs から始まるトークンでしか実行できないようです(参考issue)。そのため、本稿では触れません。

絵文字一覧の取得

Slack には emoji.list という API があるので、これを通信ライブラリの requests で実行します。

import requests
# TOKEN には Slack API のトークンを入れる
res = requests.get('https://slack.com/api/emoji.list', headers={'Authorization': f'Bearer {TOKEN}'})
emojis = res.json()['emoji']

emojis 変数は、次のような "絵文字名": "画像URL" という形式の dict 型です。また、結果には alias も含んでいます。

{
    ...
    "saiko": "https://emoji.slack-edge.com/XXXXXXX/saiko/0123456789abcdef.png",
    "shussha": "alias:syussha",  # alias のパターン
    ...
}

絵文字の URL から、画像データ(バイナリ)を取得します。

img_bytes = requests.get(url).content

闇絵文字判定

続いては、取得した画像データが、「黒文字・透過」の絵文字かどうかを判定します。

今回は、Python での画像処理に PillowNumPy を利用します。Pillow は PIL=Python Image Library のフォークで、画像処理用のライブラリです。NumPy は行列計算などに使われるライブラリです。これらライブラリを組み合わせ、取得した絵文字のバイナリ(png)を配列(高さ×幅×RGBA)に展開して操作していきます。

from io import BytesIO
from PIL import Image
import numpy as np
img = Image.open(BytesIO(img_bytes))
img_arr = np.asarray(img)  # img_arr.shape == (128, 128, 4)

透過画像かどうかの判定は簡単です。img_arr は RGBA=(赤, 緑, 青, 透明度) の 4 チャンネルのため、Aの最小値が 0 であれば透明です。

transparent = len(img_arr.shape) == 3 and img_arr.shape[2] == 4 and img_arr[:, :, 3].min() == 0

次に、暗い文字かどうかを判定します。この判定はいろいろ調整した結果、次のように判定することにしました。

gray = img_arr[:, :, :3].max(axis=2).astype(float)  # 簡易的なグレースケールとして、RGBの最大値を取ります
gray *= (img_arr[:, :, 3] / 255)  # 透過度を反映したグレー値を計算します
dark = np.percentile(gray[img_arr[:, :, 3] > 0], 80) <= 85  # 80%パーセンタイルが、85=255/3 の明るさ以下の場合、暗いと判定

シンプルなロジックですが、比較的精度良く闇絵文字を判定できています。

f:id:yamitzky:20191225003455p:plain

白背景の合成

最後に、闇絵文字と判定されたものに白背景を付与します。

white = Image.fromarray(np.ones(img_arr.shape, dtype=img_arr.dtype) * 255)  # 真っ白な画像を作る
white.paste(img, (0, 0), img.split()[3])  # 白背景に、元の画像を貼り付ける
white.save(f'result/{name}.png')

完成!

こうして、無事に闇の魔術に対抗することができました。安心して Slack をダークモードにして使えそうです。

f:id:yamitzky:20191225003505p:plain

完成した Python スクリプトは Gist に添付しましたので、ぜひ参考にしてください。

gist.github.com