L2接続とL3接続の違い

Ciscoのページがとても分かりやすかった。

違いを表にまとめておく。

レイヤー	OSIモデル	アドレス	通信単位	接続対象
L2	データリンク層	MACアドレス	フレーム	LAN内の端末
L3	ネットワーク層	IPアドレス	パケット	LAN同士

GCPとjavascriptでaccess trackerを作る（3）

backend（リクエストを受けてpub/subに流すコンポーネント）がほぼ出来上がった。

https://github.com/Kenji-H/access-tracker-backend/tree/v0.9.2

進捗

• google container engineのクラスタを立てた
• deploymentとserviceの定義を書いた
• READMEを書いた
• alpineで動かすとgcpのライブラリのロードでこけることがわかったのでfixした

動作確認

データを投げる。

$ cat test.json
{
  "userid": "610KZ4G0CBTI419Y",
  "timestamp": 1509290846000,
  "pageid": "test-page-001",
  "browser": "chrome",
  "country":"japan"
}
$ curl -H 'Content-Type:application/json' -d @test.json xxx.xxx.xxx.xxx/pv

アプリケーションのログを確認する。

$ kubectl --namespace=backend logs -f access-tracker-backend-xxxxxxxx
received request for URL: /pv
success: {"userid":"610KZ4G0CBTI419Y","timestamp":1509290846000,"pageid":"test-page-001","browser":"chrome","country":"japan"}

pub/subにデータが流れたことを確認する。

gcloud beta pubsub subscriptions pull test --auto-ack --max-messages 1000
│ {"userid":"610KZ4G0CBTI419Y","timestamp":1509290846000,"pageid":"test-page-001","browser":"chrome","country":"japan"} │ 165859404777865 │            │

次やること

• droneサーバを立てる
• unittestを書く
• CI/CDの設定をする

GCPとjavascriptでaccess trackerを作る（2）

https://github.com/Kenji-H/access_tracker/tree/v0.9.1

今日の進捗

• pub/subにデータをpublishする部分を実装した
• データをvaldationする機能を実装した
• expressに404と500のhandlerを追加した

新しく学んだこと

• node.js向けのgcp client libraryは基本非同期処理で書くようになっているが、callbackを省略するとpromiseオブジェクトを返してくれる。
• promiseオブジェクトをチェーンで繋げて書くとうまく書けそうだが、繋げたいメソッドの入出力がうまくハマらなかったり、細かい例外処理をしたかったりすると、自分でpromiseオブジェクトを作成した方が書きやすい感じがした。たぶん自分のnode.js力が低いだけで、慣れてきたらもっとうまく書けそう。
• gcpのcredentials/projectidは環境変数に入れるとすっきりする。
• jsonをバリデーションしたい場合は、jsonschemaというライブラリを使うといいらしい。
• expressにはエラーハンドラを登録することができる。404のハンドラの定義はファイルの一番下部でやらないといけない。

次やること

• gkeのクラスタを立てる
• gkeでコンテナを動かす
• プロジェクト構成を整理する（デプロイの単位は分けるつもりだったが、レポジトリ自体も分けた方がすっきりしそう）

GCPとjavascriptでaccess trackerを作る（1）

　最近GCPを使ったインフラ構築と、javascriptへの興味が高まっているので、勉強がてら何か作ってみることにした。

作りたいやつの機能

• サーバサイドはnode.jsで書く。REST APIでページビュー情報をPOSTしたり、ページビュー情報をGETしたりできる。
• POSTされた情報はGCPのpub/subにpublishされる。
• GETするときはpub/subからsubscribeする。
• サーバ自体はGCPのcontainer engineで動作する。
• クライアントサイドはReact.jsで書く。
• グラフライブラリを使ってかっこいいグラフを表示する。
• Websocketを使ってリアルタイムにグラフが更新されるようにする。

今日の進捗

node.jsでサーバを書いて、Dockerで動くようにした。

https://github.com/Kenji-H/access_tracker/tree/v0.9.0

ローカルでの動作確認方法は以下のとおり。

まず、dockerを起動。

$ cd server
$ docker build -t test .  
$ docker run --rm -p 8080:8080 test

クライアントからリクエスト送信。

$ curl localhost:8080/status
$ curl -H 'Content-Type:application/json' -d '{"userid": "610KZ4G0CBTI419Y", "timestamp": 1509130332, "url": "kenjih.com", "userAgent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36"}' localhost:8080/pv

dockerにログが出ていることを確認。

received request for URL: /status

received request for URL: /pv
userid: 610KZ4G0CBTI419Y
timestamp: 1509130332
url: kenjih.com
userAgent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36

Signal Handler in Python

PythonでSignal Handlerを書いて遊んでみた。

プログラム

Python 3で動作確認してます。

遊び方

まず上記のPythonプログラムを起動します。

$ python sample.py

プログラムを起動したら以下の表を参考にシグナルを送ってみてください。

シグナル	キーボードから	コマンドラインから
SIGINT	ctrl + c	kill -INT プロセス番号
SIGTSTP	ctrl + z	kill -TSTP プロセス番号
SIGTERM	-	kill プロセス番号
SIGKILL	-	kill -KILL プロセス番号

プログラムを停止したい場合は、SIGKILLを送ってください。
プログラムのプロセス番号は以下のコマンドで調べることができます。

$ pgrep -f "python sample.py"

単一障害点とは

まえがき

システム全体のアーキテクチャを考えるときに、「単一障害点」という概念が大事らしい。
意味は一目瞭然だけど、大切な概念っぽいので調べてみることにした。

語句の意味

その単一箇所が働かないと、システム全体が障害となるような箇所のこと。
ちなみに英語だとSingle Point of Failure（SPOF）というらしい。

単一障害点の例

コンポーネント	説明
アプリケーションサーバ	アプリケーションサーバが一つしかないと、クリティカルなエラーが発生したときにユーザがシステムを利用できなくなってしまう。よってサーバをレプリケーションしておくことでSPOFにならないようにする必要がある。
ディスクストレージ	ディスクストレージが故障するとデータの読み書きができなくなり、システム全体が利用不可になってしまう可能性がある。よってRAIDなどの冗長構成を取る必要がある。
ネットワークスイッチ	ネットワークスイッチが故障と、ネットワーク内のサーバ間の疎通ができなくなってしまう。よって冗長なスイッチとネットワーク構成によって、SPOFを解決する必要がある。

トレードオフ

SPOFを解決するためには、レプリケーションや冗長構成などが必要となるので、コストがかかる。
どこまでSPOFの排除に取り組むかはコストとのトレードオフも考えて決定しなければならない。

Perplexityとは

　自然言語処理で良く使われるメトリクスのひとつであるperplexityについて説明します。

定義

分布pに従って生成されるデータxがある。xの分布をqでモデリングする。
このとき確率モデルqのperplexityは、

と表される。

ただし、xiはpから生成したデータで、Nはxiの個数である。

直感的な説明

• 指数部はクロスエントロピーになっている
• モデルにデータを見せたときに、どれくらい"驚く"かを表す
• いいモデルほどperplexityは小さい

サンプル

サイコロの目の分布をモデリングしてみて、そのモデルのperplexityを計算してみます。
まずサイコロを用意します。サイコロは六面で各面が出る確率は一様とします。

真の分布はわからないので、30回サイコロを振ってみて観測化された結果から分布を推定してみます。このモデルのperplexityをテストデータ1,000個を使って計算してみると、6.358となります。

import numpy as np
from collections import Counter

np.random.seed(1234)


N = 1000     # test data size
M = 30       # train data size

test_data = np.random.randint(6, size=N)
train_data = np.random.randint(6, size=M)

p = Counter(test_data)    # test distribution
q = Counter(train_data)   # proposed distribution

exponent = 0
for i in range(6):
  exponent -= p[i] / N * np.log2(q[i] / M)

print (2 ** exponent)

次に訓練データを増やしてモデリングしてみます。サイコロを1,000回振ってみます（上記のコードのMを1000に変更してプログラムを実行）。
perplexityは5.998となり、モデルの良さが改善されたことがわかります。

ElastiCacheを使ってみた

　Redisのマネージドサービスを試しに使ってみた。AWSではElastiCacheというサービス名で提供されている（ElastiCacheでは、Redisの他にMemcachedも使うことができる）。

Redisクラスタの作成

• AWSコンソール> ElastiCacheダッシュボード> サブネットグループからサブネットグループを作成
• サブネットグループはRedisクラスタを作るときに指定しないといけないので、あらかじめ作っておく
• AWSコンソール> ElastiCacheダッシュボード> RedisからRedisクラスタを作成
• 練習用なのでレプリカは作らない
• Redisと同じAZにEC2を立てたいので、優先AZを指定しておく
• EC2からアクセスできるようにSGを設定しておく（今回のサンプルではdefault SGをつける）

EC2の作成

• Redisクラスタと同じVPC、subnet内にEC2を立てる
• AmazonElastiCacheFullAccessポリシーをアタッチしたロールを付与しておく
• RedisクラスタにアクセスできるようにSGを設定しておく（今回のサンプルではdefault SGをつける）

クライアントアプリ

• Pythonから使う
• pipでredisライブラリを入れる
• pip install redis
• get, set, hset, hget, hmset, hmgetとかを試してみる
• 速度がどれくらいでるか気になったので以下のプログラムで測定してみた

import time
import redis

host = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx.amazonaws.com'
port = 6379
r = redis.StrictRedis(host=host, port=port)

T = 10000

set_elapsed_times = []
for i in range(T):
    key = 'k%d' % i
    val = 'v%d' % i
    start_time = time.time()
    r.set(key, val)
    t = time.time() - start_time
    set_elapsed_times.append(t * 1e6)  # micro sec
    
get_elapsed_times = []
for i in range(T):
    key = 'k%d' % i
    val = 'v%d' % i
    start_time = time.time()
    v = r.get(key)
    t = time.time() - start_time
    get_elapsed_times.append(t * 1e6)  # micro sec

測定結果

• get/secともに、平均で250 micro secくらい
• setのmaxが13milli secくらいかかってるのが気になる

処理	GET [microsec]	SET [microsec]
mean	259.323335	241.534066
std	138.918826	232.519214
min	212.669373	206.708908
25%	234.842300	224.113464
50%	246.286392	231.504440
75%	260.114670	241.279602
max	3891.468048	13475.656509

ついでにボックスプロット（1% - 99%範囲外のデータは例外とみなして除去）も書いてみた。

expressionとstatementの違い

まえがき

Scala Schoolを読んでいて、以下のような表現に遭遇した。

Scala is highly expression-oriented: most things are expressions rather than statements.

日本語ではexpressionは「式」、statementは「文」と訳される。

両者の違い

プログラムの例を見ると言わんとすることは分かるけど、両者の定義が明確に分からなかったので調べてみた。

• 式は値を生み出す、文は何か処理をする
• 式は文の部分集合
• 式は”それは何か”を表しており、文は"何をするか"を表している
• 関数型言語は式を、命令型言語は文を使うことが多い気がする

Pythonでの例

if/else
以下のようにif/elseはstatementっぽくも書けるし、expressionっぽくも書ける。

# statement
x = 1
if x % 2 == 0:
 y = "even"
else:
 y = "odd"

# expression
"even" if x % 2 == 0 else "odd"

print
printはPython 2までは文だったが、Python3から式になった。

Python 2では文なので値は返されない。

>>> type (print ("hello"))
  File "<stdin>", line 1
    type (print ("hello"))
              ^
SyntaxError: invalid syntax

Python 3では式なので値が返される。

>>> type (print ("hello"))
hello
<class 'NoneType'>

IAMユーザにIPアドレス制限をかける

AWSのIAMユーザにIPアドレスをかける手順＆確認。

IAMユーザの作成

AWSコンソールからIAMユーザを作成する。

今回の例では、AmazonS3ReadOnlyAccessポリシーをアタッチしておく。

接続確認

コマンドラインからprofileを設定

$ aws configure --profile test-user

S3にアクセスしてみる。

$ aws --profile test-user s3 ls
2016-02-20 15:50:19 xxxxxxxxxxxxx
2016-02-20 16:10:29 yyyyyyyyyyyyy

IPアドレス制限をかける

AmazonS3ReadOnlyAccessをコピーして以下のようなポリシーを作る。（xx.xx.xx.xxのところは自分のIPアドレスを入れる）

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:Get*",
                "s3:List*"
            ],
            "Resource": "*",
            "Condition": {
                "IpAddress": {
                    "aws:SourceIp": [
                        "xx.xx.xx.xx/32"
                    ]
                }
            }
        }
    ]
}

AmazonS3ReadOnlyAccessをデタッチして、上記で作成したポリシーを代わりにアタッチする。

指定したアドレスから接続確認

S3にアクセスできることを確認。

$ aws --profile test-user s3 ls
2016-02-20 15:50:19 xxxxxxxxxxxxx
2016-02-20 16:10:29 yyyyyyyyyyyyy

指定したアドレス以外から接続

S3にアクセスできないことを確認。

$ aws --profile test-user s3 ls
A client error (AccessDenied) occurred when calling the ListBuckets operation: Access Denied

オントロジーとタクソノミーの違い

オントロジー（Ontology）とは

• type, properties, relationshipの集合から構成される世界を記述するモデルのこと。
• Grakn Knowledge Modelというモデルではオントロジーを以下の4つのコンポーネントで記述する。
• entity: もの。例: 人、男、女
• relation: entity同士の関連。例: 親子関係
• role: relationにおける役割。例: "結婚"というrelationには、"夫"、"妻"というroleがある
• resource: entity/relationに紐づく属性。 例:名前、日付

タクソノミー（Taxonomy）とは

• entityを親子関係を使って階層的に並べて表現するモデル。
• 例: 植物 - 被子植物 - 単子葉類 - ユリ目 - ユリ科 - チューリップ属

両者の違い

• Ontologyの方がTaxonomyより複雑な表現をすることができる。
• Taxonomyは木構造、Ontologyはグラフ構造になっているとイメージしておくとよさそう。
• Taxonomyはentityの分類や類似度を測るときに使われるのに対して、Ontologyは知識表現に使われる。Ontologyを使って知識を表現しておくと、マシンを使って新しい知識を推論することができる。

参考
https://www.quora.com/Whats-the-difference-between-an-ontology-and-a-taxonomy
https://blog.grakn.ai/what-is-an-ontology-c5baac4a2f6c

KMSを使ってみた

はじめに

　AWSのKMS（Key Management Service）を使ってみた。
KMSを使うと以下のようなことができて嬉しい。

• ソースコードに載せたくない情報（認証情報など）を暗号化できる
• 暗号化した鍵の管理をAWS側でやってくれる

やってみたこと

1. キーの作成

AWSコンソール画面 > IAM > 暗号化キー > キーの作成

からキーを作っておく

2. キーのAmazon Resource Nameをメモ

1.で作成したキーのリンクをクリックして、arn:aws:kms:xxxxxxxxxxxxxxのところをメモっておく

3. 検証に使うec2を起動

KMSにアクセスできるロールを適用したec2の起動する

4. 3.のロールから1.のキーのアクセスを許可

AWSコンソール画面 > IAM > 暗号化キー > キーユーザー

4.の操作はec2ロールにアタッチしたポリシーに応じて必要だったり、不要だったりする。

• ec2にPowerUserAccessポリシーをつけていれば、3.の操作は不要。
• AWSKeyManagementServicePowerUserポリシーしかつけてない場合は、3.の操作が必要。

5. 検証用Pythonコードを実行

注意事項

キーを作成したリージョンとキーを利用するリージョンが異なると以下のようなエラーがでるので注意。

botocore.errorfactory.NotFoundException: An error occurred (NotFoundException) 
when calling the Encrypt operation: Invalid arn

論文英語

　論文を書くときに使うとかっこ良さそうな英語表現をまとめたページ。

adaptively: 適応的に
It encodes the input sentence into a sequence of vectors and chooses a subset of these vectors adaptively while decoding the translation.（引用元）

aforementioned: 前述の
In addition to the aforementioned relevance criterion, ...（引用元）

exhibit: 示す、提示する
the relative difference between the two algorithms exhibits a relatively large variance over different domains.（引用元）

heterogeneous: 異種の、異質の

the adopted L2R framework makes room for a seamless integration of heterogeneous models and data sources, by merging all of them into the  nal scoring model in the form of additional features.（引用元）

homogeneous: 同種の、同質の
the novel information can be homogeneously fed into the L2R engine in the form of additional features.（引用元）

outperform: 〜より性能がすぐれている
It is clear from the table that in all the cases, the proposed RNNsearch outperforms the conventional RNNencdec. （引用元）

qualitative analysis: 定性的な分析
We also present some qualitative analysis of the output from our models. （引用元）

quantitative results: 定量的な結果
In conjunction with the quantitative results presented already, ... （引用元）

to our knowledge: 我々が知る限り
To our knowledge, this is the first time a multi-objective recommendation problem is addressed within the learning-to-rerank framework. （引用元）

[Blog Reading] Hiring SREs at LinkedIn

　SREの採用プロセスに関する話。SREに限らずエンジニアの採用プロセスについてのいい話が書いてある。

https://engineering.linkedin.com/blog/2017/07/hiring-sres-at-linkedin

• 人を雇いたいときは、その人に満たしてほしい特定のニーズがある
• その特定のニーズを満たすために必要なスキルはなにか
• 採用試験ではそのスキルだけをテストする
• まず電話・オンラインで試験をする（スケールしやすい）
• オンサイトでの試験は受験者/採用者ともにコストがかかる
• 複数回のオンライン試験をパスした有望な求職者のみをオンサイト試験に招待する
• オンサイト試験ではオンライン試験でやらなかったことをやる

Nat Gatewayが意外と高かった

最近AWSでネットワークの勉強をしているのだけど、請求書を見てみたら予想外にコストがかかっていた。

EC2は都度停止しているはずなのに何故かなと思って明細を見ていたら、Nat Gatewayだった。

1時間あたりのコストがt2.microインスタンスの4倍近くかかるらしい。VPC自体は無料で作れるので、NATも無料だろうと思って油断していた。

（追記）
マネージドなNATがなかった頃は、EC2インスタンスを立ててNAT用のマシンにしていたらしい。NATが落ちるとprivate networkのマシンからインターネットが見れなくなるので冗長化したり、パフォーマンスを考えてそこそこいいEC2を使ったりしないといけなかったらしく、それを考えると妥当な値段だということが分かった。
ただし、個人で学習用に使う場合はやはり高いので、動作確認ができたらさっさと削除しましょう。

並行と並列の違い

並行（concurrent）処理とは、複数のタスクを非同期で同時に実行すること。
並列（parallel）処理とは、複数のCPUを使って複数のタスクを同時に実行すること。

つまり、並行の方が広い概念で、並列処理は並行処理の一種。

並行処理にはいくつかの実現方法が考えられるが、複数のCPUを使って物理的に同時に複数の命令を走らせているものを並列処理と呼ぶ。

ということで、場面によって使い分けた方がいいかもしれない。

• 複数のサーバを立てて並列処理をする
• マルチコアCPUで並列処理をする
• シングルコアマシン上でマルチスレッドを使って並行処理をする
• コルーチンで並行処理をする

並行の方が広い概念なので、並行処理と言っておけば間違いは少なそうだが、物理的に同時に実行するということを強調したい場合は並列と言った方がいいかもしれない。

並列処理

並列ではない並行処理

[Blog Reading] Neflix Platform Engineering — we’re just getting started

　エンジニアリングの世界に終わりはない、常に新しいチャレンジが待っているという内容のブログ。

Neflix Platform Engineering — we’re just getting started

• 2, 3年ほど前にKafkaを使うようになった
• 最近はApache Flinkをストリーム処理に使うようになった
• AMI/VMからcontainer technologyに移行
• 数100のマイクロサービスが動いている
• Chaos Monkey（NetflixのOSS）からChaos Engineeringという規範が生まれた

[Blog Reading] Building a Real-Time Streaming ETL Pipeline in 20 Minutes

Confluentのブログを読んだ。
最近KafkaまわりのOSSにプルリク送ってみたりして、Confluent熱が個人的に高まっている。

Building a Real-Time Streaming ETL Pipeline in 20 Minutes

以下、感想・考えてみたことなど。

• Kafkaってなんなのか一言で説明するのは難しいが、このブログにあるように「分散ストリーミング基盤」というのが良さそう。
• Streaming ETLという概念がやばい。Batch ETLでやっているような集計処理をバックグラウンドでリアルタイムに実行できるやつ。
• ブログの下の方で引用されているが、簡単なサンプルアプリがあるので見てみるとイメージがわく
• 「複数のアプリが同じデータソースを参照していると密結合になる」みたいな話を聞いたことがあったが、その感覚が分かった。例えば、アプリAがデータ生成するテーブルAがあったとして、これをみたい他のアプリB, C, D, ..があったとする。アプリB, C, D, ..からテーブルAを直接参照すると以下のような問題が起こる。
• 参照者が増えてくると、テーブルAのアクセス負荷が増える（Producer:Consumer = 1:N）
• アプリAが自由にテーブルAのスキーマを変更できない（テーブルAの変更は参照側のアプリに影響を与える）
• 参照で集計処理をしたい場合は、日次バッチになってしまう（負荷が高いのでリアルタイムで集計できない）

VagrantのゲストOS間でホスト名参照

 Vagrantfileでホスト名を指定してるのに何故か参照できなかった。

# -*- mode: ruby -*-
# vi: set ft=ruby :

Vagrant.configure("2") do |config|
  config.vm.box = "ubuntu15.04"
  config.ssh.insert_key = false
  config.vm.define "kafka-base" do |server|
    server.vm.network "private_network", ip: "192.168.33.11"
    server.vm.hostname = "kafka-base"
  end
  config.vm.define "kafka-connector" do |server|
    server.vm.network "private_network", ip: "192.168.33.12"
    server.vm.hostname = "kafka-connector"
  end
end

vagrant@kafka-base:~$ hostname
kafka-base
vagrant@kafka-base:~$ ping kafka-connector
ping: unknown host kafka-connector
vagrant@kafka-base:~$ cat /etc/hosts
127.0.0.1 kafka-base  kafka-base
127.0.0.1 localhost
127.0.1.1 vagrant-ubuntu-trusty.vagrantup.com vagrant-ubuntu-trusty
::1     localhost ip6-localhost ip6-loopback
ff02::1 ip6-allnodes
ff02::2 ip6-allrouters

vagrant-hostsというプラグインを入れてみると解決した。

$ vagrant plugin install vagrant-hosts

以下のようにserver.vm.provisionの行を追加すると、ホスト名が設定され、さらにゲストOSの/etc/hostsに仮想マシンのホスト名情報が追記される。

# -*- mode: ruby -*-
# vi: set ft=ruby :

Vagrant.configure("2") do |config|
  config.vm.box = "ubuntu15.04"
  config.ssh.insert_key = false

  config.vm.define "kafka-base" do |server|
    server.vm.network "private_network", ip: "192.168.33.11"
    server.vm.provision :hosts, :sync_hosts => true
  end
  config.vm.define "kafka-connector" do |server|
    server.vm.network "private_network", ip: "192.168.33.12"
    server.vm.provision :hosts, :sync_hosts => true
  end
end

以下のように動作確認してみると、ホスト 名でゲストOS間の通信ができることが分かる。

vagrant@kafka-base:~$ hostname
kafka-base
vagrant@kafka-base:~$ ping kafka-connector
PING kafka-connector (192.168.33.12) 56(84) bytes of data.
64 bytes from kafka-connector (192.168.33.12): icmp_seq=1 ttl=64 time=0.470 ms
64 bytes from kafka-connector (192.168.33.12): icmp_seq=2 ttl=64 time=0.348 ms
^C
--- kafka-connector ping statistics ---
2 packets transmitted, 2 received, 0% packet loss, time 1002ms
rtt min/avg/max/mdev = 0.348/0.409/0.470/0.061 ms
vagrant@kafka-base:~$ cat /etc/hosts
127.0.0.1 localhost
127.0.1.1 kafka-base
192.168.33.11 kafka-base
192.168.33.12 kafka-connector

Vagrantで共通のssh private keyを使う

　Vagrantで複数の仮想マシンを立ち上げて、仮想マシン間でsshの秘密鍵を共有したい場合がある。Vagrant 1.8でのデフォルトの設定ではマシンごとに異なる秘密鍵が生成される。

# -*- mode: ruby -*-
# vi: set ft=ruby :

Vagrant.configure("2") do |config|
  config.vm.box = "ubuntu15.04"
  config.vm.define "kafka-base" do |server|
    server.vm.network "private_network", ip: "192.168.33.11"
  end
  config.vm.define "kafka-connector" do |server|
    server.vm.network "private_network", ip: "192.168.33.12"
  end
end

秘密鍵情報を確認。仮想マシンごとに別の秘密鍵が生成されていることが分かる。

$ vagrant ssh-config
Host kafka-base
  HostName 127.0.0.1
  User vagrant
  Port 2200
  UserKnownHostsFile /dev/null
  StrictHostKeyChecking no
  PasswordAuthentication no
  IdentityFile /Users/kenjih/work/vagrant_ansible_kafka/vagrant/.vagrant/machines/kafka-base/virtualbox/private_key
  IdentitiesOnly yes
  LogLevel FATAL

Host kafka-connector
  HostName 127.0.0.1
  User vagrant
  Port 2201
  UserKnownHostsFile /dev/null
  StrictHostKeyChecking no
  PasswordAuthentication no
  IdentityFile /Users/kenjih/work/vagrant_ansible_kafka/vagrant/.vagrant/machines/kafka-connector/virtualbox/private_key
  IdentitiesOnly yes
  LogLevel FATAL

　しかし、デプロイ自動化を行う場合など、共通の秘密鍵を使えると便利なことが多い。そのような場合は以下のようにconfig.ssh.insert_key = falseという設定を追加すればよい。

# -*- mode: ruby -*-
# vi: set ft=ruby :

Vagrant.configure("2") do |config|
  config.vm.box = "ubuntu15.04"
  config.ssh.insert_key = false
  config.vm.define "kafka-base" do |server|
    server.vm.network "private_network", ip: "192.168.33.11"
  end
  config.vm.define "kafka-connector" do |server|
    server.vm.network "private_network", ip: "192.168.33.12"
  end
end

秘密鍵情報を確認してみると、共通の秘密鍵を利用できることが分かる。

kenjih$ vagrant ssh-config
Host kafka-base
  HostName 127.0.0.1
  User vagrant
  Port 2200
  UserKnownHostsFile /dev/null
  StrictHostKeyChecking no
  PasswordAuthentication no
  IdentityFile /Users/kenjih/.vagrant.d/insecure_private_key
  IdentitiesOnly yes
  LogLevel FATAL

Host kafka-connector
  HostName 127.0.0.1
  User vagrant
  Port 2201
  UserKnownHostsFile /dev/null
  StrictHostKeyChecking no
  PasswordAuthentication no
  IdentityFile /Users/kenjih/.vagrant.d/insecure_private_key
  IdentitiesOnly yes
  LogLevel FATAL

VagrantとAnsibleでKafka環境をつくる（８）

　Vagrant、Ansibleで作ったKafka環境で動作確認をしてみる。
以下の操作はすべてプロジェクトのrootディレクトリから実行するものとします。

まずvagrantで仮想マシンを起動する。

$ cd vagrant
$ vagrant up

続いてansibleでplaybookを適用する。

$ cd ansible
$ ansible-playbook site.yml 

まずはrest proxyと疎通確認を行う。

$ curl 192.168.33.11:8082/topics
["_schemas"]

rest proxy経由でKafkaにデータを送ってみる。

$ curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" \
 --data '{"records":[{"value":{"name": "testUser"}}]}' \
      "192.168.33.11:8082/topics/jsontest"
{"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

consumerを作成し、先ほど作成したjsontestトピックをサブスクライブする。

$ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" -H "Accept: application/vnd.kafka.v2+json" \
--data '{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}' \
http://192.168.33.11:8082/consumers/my_json_consumer
{"instance_id":"my_consumer_instance","base_uri":"http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance"}

$ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" --data '{"topics":["jsontest"]}' \
http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance/subscription

もう1つメッセージを送ってみる。

$ curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" \
 --data '{"records":[{"value":{"name": "testUserXXX"}}]}' \
      "192.168.33.11:8082/topics/jsontest"
{"offsets":[{"partition":0,"offset":1,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

メッセージをコンシュームする。

$ curl -X GET -H "Accept: application/vnd.kafka.json.v2+json" \       
http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance/records                               
[{"key":null,"value":{"name":"testUser"},"partition":0,"offset":0,"topic":"jsontest"},{"key":null,"value":{"name":"testUserXXX"},"partition":0,"offset":1,"topic":"jsontest"}]

データの送受信を確認できたので、テスト用のコンシューマーを削除する。

$ curl -X DELETE -H "Accept: application/vnd.kafka.v2+json" \
      http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance

VagrantとAnsibleでKafka環境をつくる（７）

　前回マニュアルで試したsystemdからconfluentコンポーネントを起動するための設定をansible化した（成果物）。利用するコンポーネントは以下の4つなので、それぞれのコンポーネント別にroleを作った。

　serviceの設定スクリプトは、Jinja2を使ってtemplatesディレクトリ配下に置いている。taskではtemplatesに置かれたテンプレートを/etc/systemd/system/配下に格納する。設定が変更されると、handlerで指定したようにserviceが起動される。

　以下にzookeeper roleの場合のサンプルを載せておく。

ansible/roles/zookeeper/templates/zookeeper.service.j2

[Unit]
Description=confluent platform zookeeper
After=network.target

[Service]
ExecStart=/usr/bin/zookeeper-server-start /etc/kafka/zookeeper.properties
ExecStop=/usr/bin/zookeeper-server-stop

[Install]
WantedBy=multi-user.target

ansible/roles/zookeeper/tasks/main.yml

- name: zookeeper systemd script
  template:
    src: zookeeper.service.j2
    dest: /etc/systemd/system/zookeeper.service
    owner: root
    group: root
    mode: 644
  notify: start zookeeper

ansible/roles/zookeeper/handlers/main.yml

- name: start zookeeper
  service: name=zookeeper state=started enabled=yes

VagrantとAnsibleでKafka環境をつくる（６）

　zookeeperをサービス登録して、systemdから起動させるようにしてみた。
Ansible化はまだ出来ていないが、とりあえず手動で設定＆起動できた。

以下のようなファイルを作っておく。

vagrant@vagrant-ubuntu-trusty:~$ cat /etc/systemd/system/zookeeper.service 
[Unit]
Description=confluent platform zookeeper
After=network.target

[Service]
ExecStart=/usr/bin/zookeeper-server-start /etc/kafka/zookeeper.properties
ExecStop=/usr/bin/zookeeper-server-stop

[Install]
WantedBy=multi-user.target

After=network.targetとすることで、ネットワーク起動後にサービスを開始させるという意味になる。
WantedByのところには、ランレベルを設定する。
multi-user.targetとするとマルチユーザモードで使用されるサービスとなる。

ちなみに、targetの一覧は以下のようにして参照できる。

vagrant@vagrant-ubuntu-trusty:~$ systemctl list-units --type target
UNIT                  LOAD   ACTIVE SUB    DESCRIPTION
basic.target          loaded active active Basic System
cryptsetup.target     loaded active active Encrypted Volumes
getty.target          loaded active active Login Prompts
graphical.target      loaded active active Graphical Interface
local-fs-pre.target   loaded active active Local File Systems (Pre)
local-fs.target       loaded active active Local File Systems
multi-user.target     loaded active active Multi-User System
network-online.target loaded active active Network is Online
network.target        loaded active active Network
nfs-client.target     loaded active active NFS client services
paths.target          loaded active active Paths
remote-fs-pre.target  loaded active active Remote File Systems (Pre)
remote-fs.target      loaded active active Remote File Systems
rpcbind.target        loaded active active RPC Port Mapper
slices.target         loaded active active Slices
sockets.target        loaded active active Sockets
swap.target           loaded active active Swap
sysinit.target        loaded active active System Initialization
time-sync.target      loaded active active System Time Synchronized
timers.target         loaded active active Timers

サービスとして認識しているか確認。

vagrant@vagrant-ubuntu-trusty:~$ sudo systemctl list-unit-files --type=service | grep zookeeper
zookeeper.service                      disabled

サービス有効化。

vagrant@vagrant-ubuntu-trusty:~$  sudo systemctl enable zookeeper
Created symlink from /etc/systemd/system/multi-user.target.wants/zookeeper.service to /etc/systemd/system/zookeeper.service.

サービス起動＆確認。

vagrant@vagrant-ubuntu-trusty:~$ sudo systemctl start zookeeper

vagrant@vagrant-ubuntu-trusty:~$ sudo systemctl status zookeeper
● zookeeper.service - confluent platform zookeeper
   Loaded: loaded (/etc/systemd/system/zookeeper.service; enabled; vendor preset: enabled)
   Active: active (running) since Mon 2017-06-05 14:22:49 GMT; 16s ago

デフォルトポートでlistenしているか念のため確認。

vagrant@vagrant-ubuntu-trusty:~$ pgrep -f zookeeper
4542
vagrant@vagrant-ubuntu-trusty:~$ sudo lsof -a -i -p 4542
COMMAND  PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    4542 root   98u  IPv6  25718      0t0  TCP *:56431 (LISTEN)
java    4542 root  109u  IPv6  25729      0t0  TCP *:2181 (LISTEN)

VagrantとAnsibleでKafka環境をつくる（５）

　前回Javaが入ったので、Kafka環境を入れてみた（成果物）。
KafkaはConfluent Platformを使う。手動でインストールすると以下のようになる。

# install confluent public key
$ wget -qO - http://packages.confluent.io/deb/3.2/archive.key | sudo apt-key add -

# add confluent repository
$ sudo add-apt-repository "deb [arch=amd64] http://packages.confluent.io/deb/3.2 stable main"

# update apt
$ sudo apt-get update 

# install confluent platform
$ sudo apt-get install confluent-platform-2.11

上のインストール作業を行うplaybookは以下のようになる。

[ansible/roles/common/tasks/install_confluent.yml]

- name: install confluent public key
  apt_key: url="http://packages.confluent.io/deb/{{ confluent_repo_version }}/archive.key" state="present"

- name: add confluent repository
  apt_repository: repo="deb [arch=amd64] http://packages.confluent.io/deb/{{ confluent_repo_version }} stable main"

- name: update apt
  apt: update_cache=true

- name: install confluent platform
  apt: "name=confluent-platform-2.11={{ confluent_package_version }} state=present"

{{ XXX }}のところは変数になっている。変数の値は以下のようにvarsディレクトリ内のファイルで定義できる。

[ansible/roles/common/vars/main.yml]

confluent_repo_version: 3.2
confluent_package_version: 3.2.1-1

VagrantとAnsibleでKafka環境をつくる（４）

　Oracle Java1.8を入れるplaybookを書いた（成果物）。

　playbookはroleごとにまとめるのがbest practiceらしい。サーバの役割毎（common, web, dbなど）にroleを設定しているサンプルが多いので、それにならって、common, zookeeper, kafka, schema-registry, rest-proxyというroleをつくろうと思う。

　今回書いたのはJavaのところだけなので、common roleにtaskを書いた。

作成したplaybookのsyntaxチェック。

$ ansible-playbook roles/common/tasks/main.yml --syntax-check

playbook: roles/common/tasks/main.yml

playbookを実行する。

$ ansible-playbook roles/common/tasks/main.yml

PLAY [all] **************************************************************************

TASK [Gathering Facts] **************************************************************
ok: [192.168.33.11]

TASK [Install add-apt-repostory] ****************************************************
changed: [192.168.33.11]

TASK [Add Oracle Java Repository] ***************************************************
changed: [192.168.33.11]

TASK [Accept Java 8 License] ********************************************************
changed: [192.168.33.11]

TASK [Install Oracle Java 8] ********************************************************
changed: [192.168.33.11] => (item=[u'oracle-java8-installer', u'ca-certificates', u'oracle-java8-set-default'])

PLAY RECAP **************************************************************************
192.168.33.11              : ok=5    changed=4    unreachable=0    failed=0

仮想マシンにJava 1.8がインストールされたことを確認。

$ vagrant ssh
vagrant@vagrant-ubuntu-trusty:~$ java -version
java version "1.8.0_131"
Java(TM) SE Runtime Environment (build 1.8.0_131-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.131-b11, mixed mode)

VagrantとAnsibleでKafka環境をつくる（３）

　Ansibleのplaybookを書こうとしたが、

• まずはマニュアルで環境を作ってみてうまくいったらplaybookを書きたい
• マニュアルでいろいろ試行錯誤して環境汚してしまうのは避けたい

ということで何か便利なツールがないか探していたところ、saharaというvagrantのプラグインを見つけた。

saharaを使うと、仮想マシンの環境をサンドボック状態にしてコミット/ロールバックを行うことができる。

以下実行例。

プラグインをインストール。

$ vagrant plugin install sahara
$ vagrant sandbox -h
Usage: vagrant sandbox  []

Available subcommands:
     commit
     off
     on
     rollback
     status

sandboxモードを有効にする。

$ vagrant sandbox on

$ vagrant sandbox status
[default] Sandbox mode is on

sandboxモードの状態で仮想マシンに何かインストールしてみる。

$ vagrant ssh default

vagrant@vagrant-ubuntu-trusty:~$ tmux
The program 'tmux' is currently not installed. You can install it by typing:
sudo apt-get install tmux

vagrant@vagrant-ubuntu-trusty:~$ sudo apt-get install tmux

vagrant@vagrant-ubuntu-trusty:~$ tmux -V
tmux 1.9

ロールバックする。

$ vagrant sandbox rollback

上記のインストールはロールバックされる。

$ vagrant ssh 

vagrant@vagrant-ubuntu-trusty:~$ tmux
The program 'tmux' is currently not installed. You can install it by typing:
sudo apt-get install tmux

VagrantとAnsibleでKafka環境をつくる（２）

Vagrantで作った仮想マシンにAnsibleから接続できるようになった（成果物）。

まず、仮想マシンにsshするときの秘密鍵の場所を調べておく。

kenjih$ vagrant ssh-config
Host default
  HostName 127.0.0.1
  User vagrant
  Port 2222
  UserKnownHostsFile /dev/null
  StrictHostKeyChecking no
  PasswordAuthentication no
  IdentityFile /Users/kenjih/work/vagrant_ansible_kafka/vagrant/.vagrant/machines/default/virtualbox/private_key
  IdentitiesOnly yes
  LogLevel FATAL

ユーザと秘密鍵の場所をオプション指定するとansibleで仮想マシンに接続できる。

kenjih$ ansible -i provisioning/hosts all -m ping -u vagrant --private-key=../vagrant/.vagrant/machines/default/virtualbox/private_key
192.168.33.11 | SUCCESS => {
    "changed": false, 
    "ping": "pong"
} 

が、毎回オプション書くのは面倒なのでconfigファイルを書くといいらしい。 以下のようなファイルをカレントディレクトリに作っておく。

$ cat ansible.cfg
[defaults]
hostfile = provisioning/hosts
remote_user = vagrant
private_key_file = /Users/kenjih/work/vagrant_ansible_kafka/vagrant/.vagrant/machines/default/virtualbox/private_key

すると、オプションなしで接続できる。

$ ansible all -m ping          
192.168.33.11 | SUCCESS => {
    "changed": false, 
    "ping": "pong"
}

接続できたので、ansibleから任意のコマンドを実行してみる。

kenjih$ ansible all -a 'whoami'
192.168.33.11 | SUCCESS | rc=0 >>
vagrant

kenjih$ ansible all -a 'uname -a'
192.168.33.11 | SUCCESS | rc=0 >>
Linux vagrant-ubuntu-trusty 3.19.0-15-generic #15-Ubuntu SMP Thu Apr 16 23:32:37 UTC 2015 x86_64 x86_64 x86_64 GNU/Linux

Distributed Code Jam Round 1 2017 E. query_of_death

問題

GetValue(i)を呼ぶとi番目の値（0 or 1）が返ってくる。

ただし、あるiを使ってGetValue(i)を呼ぶとシステムが死んでしまう（他のノードに対しては影響はない）。以下ではこのiをi_qodと呼ぶ。

i = 1 〜 NまでのGetValue(i)の和を求めよ。

制約

N <= 10^8

ノード数: 100

解法

semiexp.さんの解法が綺麗。

まずノードをmasterとslaveに分ける。

masterは計算する範囲をslaveに渡す。slaveは与えられた部分問題をといてmasterに返却する。1回目のbatchで1つのslaveが死ぬが、i_qodを含む範囲が1/99に絞られる。続いて2回目のbatchでも1つのslaveが死ぬが、i_qodを含む範囲がさらに1/98に絞られる...というふうに分割統治的に解ける。見ればわかるけど、自分じゃ思いつかない。

ソースコード

// DCJ templates begin
template <typename T>
void PutStruct(int target, const T &v) {
  char *p = (char *) &v;
  for (int i = 0; i < sizeof(T); i++) {
    PutChar(target, p[i]);
  }
}

template <class T>
T GetStruct(int source) {
  char buf[sizeof(T)];
  for (int i = 0; i < sizeof(T); i++) {
    buf[i] = GetChar(source);
  }
  return *((T *)buf);
}
// DCJ templates end

int calc(pair<int, int> pr) {
  int l, r;
  tie(l, r) = pr;

  if (l == r) return 0;
  if (r - l == 1) return GetValue(l);

  int sum = 0;
  for (int i = l; i < r; i++)
    sum += GetValue(i);

  int ck = 0;
  for (int i = 0; i < 50; i++)
    ck += GetValue(l);

  if (ck != 0 && ck != 50)
    return -1;
  return sum;
}

int main() {
  int rank = MyNodeId();
  int NN = NumberOfNodes();
  
  if (!rank) {
    long long L = 0;
    long long R = GetLength();
    set<int> alive;
    int sum = 0;
    
    for (int i = 1; i < NN; i++)
      alive.insert(i);
    
    for (;;) {
      bool update = false;

      vector<int> vs(ALL(alive));
      for (int i = 0; i < vs.size(); i++) {
        int l = L + i * (R - L) / vs.size();
        int r = L + (i + 1) * (R - L) / vs.size();
        PutStruct(vs[i], make_pair(l, r));
        Send(vs[i]);
      }

      for (int i = 0; i < vs.size(); i++) {
        Receive(vs[i]);
        int s = GetInt(vs[i]);
        if (s == -1) {
          update = true;
          int l = L + i * (R - L) / vs.size();
          int r = L + (i + 1) * (R - L) / vs.size();
          L = l, R = r;
          PutStruct(vs[i], make_pair(-1, -1));
          Send(vs[i]);
          alive.erase(vs[i]);
        } else {
          sum += s;
        }
      }
      if (!update)
        break;
    }
    
    for (auto &x : alive) {
      PutStruct(x, make_pair(-1, -1));
      Send(x);
    }
    
    cout << sum << endl;
  } else {
    for (;;) {
      Receive(0);
      auto pr = GetStruct<pair<int, int> >(0);
      if (pr.first == -1)
        break;
      int s = calc(pr);
      PutInt(0, s);
      Send(0);
    }
  }

  return 0;
}

VagrantとAnsibleでKafka環境をつくる（１）

　まずはVagrantで仮想マシンの作成、起動、停止をするところから。

boxの確認

kenjih$ vagrant box list
centos6     (virtualbox, 0)
ubuntu14.04 (virtualbox, 0)

boxの追加

kenjih$ vagrant box add ubuntu15.04 https://github.com/kraksoft/vagrant-box-ubuntu/releases/download/15.04/ubuntu-15.04-amd64.box

再びBoxの確認（ubuntu15.04が追加されている）

kenjih$ vagrant box list
centos6     (virtualbox, 0)
ubuntu14.04 (virtualbox, 0)
ubuntu15.04 (virtualbox, 0)

初期化

kenjih$ vagrant init ubuntu15.04

仮想マシン起動

kenjih$ vagrant up

仮想マシンへのログイン

kenjih$ vagrant ssh
vagrant@vagrant-ubuntu-trusty:~$
vagrant@vagrant-ubuntu-trusty:~$ uname -a
Linux vagrant-ubuntu-trusty 3.19.0-15-generic #15-Ubuntu SMP Thu Apr 16 23:32:37 UTC 2015 x86_64 x86_64 x86_64 GNU/Linux

仮想マシン状態確認（状態がrunningになっている）

kenjih$ vagrant status
default                   running (virtualbox)

仮想マシン停止

kenjih$ vagrant halt

再び仮想マシン状態確認（状態がpoweroffになっている）

kenjih$ vagrant status
default                   poweroff (virtualbox)

Pythonのデフォルト値の罠

　Pythonのデフォルト値ではまったところがあったのでメモ。以下のプログラムを例に考える。

def add_key_value_to_dict(k, v, d={}):
  d[k] = v;
  return d

if __name__ == "__main__":
  x = add_key_value_to_dict("aa", "bb", {"xx": "yy"})
  print (x)

  y = add_key_value_to_dict("hoge", "fuga")
  print (y)
  
  z = add_key_value_to_dict("foo", "bar")
  print (z)

これを実行すると以下のような結果が出力される。

{'aa': 'bb', 'xx': 'yy'}
{'hoge': 'fuga'}
{'hoge': 'fuga', 'foo': 'bar'}

1行目と2行目の結果は予想どおりだが、3行目の結果はPythonに詳しくない人には意外かもしれない。

実はPythonのデフォルト値はdefステートメントが実行されるときだけ評価される。つまりデフォルト値がmutableな値の場合、関数内で実行した変更は副作用をもたらすことになる。上の例では、dict型の引数dのデフォルトは最初は{}だが、そのあとyを計算するときに{'hoge': 'fuga'}に更新され、さらにzの計算時に{'hoge': 'fuga', 'foo': 'bar'}に更新される。

デフォルト値にmutableな値を指定すると予想しない挙動になることがあるため、以下のようにimmutableな値を使用するとよい。

def add_key_value_to_dict(k, v, d=None):
  if d is None:
    d = {}
  d[k] = v;
  return d

if __name__ == "__main__":
  x = add_key_value_to_dict("aa", "bb", {"xx": "yy"})
  print (x)

  y = add_key_value_to_dict("hoge", "fuga")
  print (y)
  
  z = add_key_value_to_dict("foo", "bar")
  print (z)

上のコードの実行結果は以下のとおり。

{'aa': 'bb', 'xx': 'yy'}
{'hoge': 'fuga'}
{'foo': 'bar'}

Kさんの教え

　半年間一緒に働いたエンジニアのKさんの教え。

おれたちはソフトウェアを作っている
ハードウェアを作る場合は、綿密な計画、設計無しにものづくりを始めてはいけない。作り直しのコストが非常に大きいからね。でも、おれたちはソフトウェアを作っている。綿密な計画、設計をしているうちに作っちゃった方が早いよね。

必要以上に複雑にするな
ソフトウェアは使ってもらいながら進化していくものなので、最初から複雑にしすぎてはいけない。絶対必要なものに集中してとにかくシンプルなものからスタートせよ。

変更は小さく頻繁に
ソフトウェアの変更は細かく、高頻度に行うべき。一度に多くのものを変えようとするのはよくない。簡単な機能なのにすぐに変更、リリースができないなんて論外。

学びこそ至福の喜び
常に学び続けろ。新しいことを学ぶのがエンジニアの大きな喜びの一つだ。学ぶことをやめてしまったエンジニアに価値はない。

おまえはどうしたいのか
あの人がこう言った、あのチームの人たちはこう言っている、じゃない。
お前はどうしたいんだ？それが一番大事なことだ。

一心不乱に夢中になること
何かに夢中になっているヤツのまわりにいると、こっちもなんだか楽しい気分になってくるぜ。おれの会社の創業者もそうさせてくれるやつだった。

セキュリティと生産性のトレードオフ
セキュリティを厳しくしすぎると生産性が著しく低下する。両者はトレードオフの関係にあるけど、日本では生産性の問題が軽視される傾向にあるみたいだ。

（番外編）マラソン
マラソンはいろんなことを教えてくれる。精神状態をいかにコントロールするか、ペース配分をどうするかなど。

（番外編）やばたん
YABATAN知らないのか？日本語だぜ。若い女の子のことのこと分かってないんじゃないか？

便利な英語表現: To our best knowledge

　論文を読んでいるとよく出てくる表現。「我々が知るかぎり〜だ」と言うときに使える。

To our best knowledge, this is the first time that ....
我々が知るかぎり、....したのはこれが初めてだ。

To our best knowledge, our model is better than all previously published methods.
我々が知るかぎり、我々の手法は過去に発表された他の手法より優れている。

はじパタ第5章 k最近傍法（kNN法）

　平井有三先生の「はじめてのパターン認識」第5章を読んだ。
kNN自体はもちろん知っていたが、理論的な考察についてはほとんど知らなかったので勉強になった。

面白かったところ

• 式を使ったボロノイ図の定義
• kNNとベイズ誤り率の関係
• kNNにおける次元の呪いの話

理解度チェックリスト

• 最近傍法とは？
• k最近傍法とは？
• ボロノイ図とは？
• kNNを使うときに問題となる次元の呪いについて説明せよ。
• 予測データと学習データの距離はどうなる？
• kNNの計算量は？
• kNNの計算量を減らすための工夫をいくつか説明せよ。

便利な英語表現: get back to

「あとで〜するね」というときに使える便利なフレーズ。

ミーティングなどで「あとで確認して連絡します。」という場合は、

I'll get back to you later.

「金曜までに具体的な数字を教えてくれますか？」とお願いするときは

Can you get back to me with some figures by Friday?

何かを注文しようとして、その場で即決できず「ちょっと考えてあとで連絡するよ」と言いたいときは

Thank you. Let me think about it and get back to you.

単位超球から一様サンプリング

　誰もが一度はモンテカルロ法を使って円周率を計算してみたことがあると思います。二次元の場合は、点が円の中に入ったり、入らなかったりするのですが、次元が大きくなると円の中に入る頻度が極端に少なくなります。
　
　じゃあ、もし高次元の単位超球内に一様分布する点をサンプリングしたくなったらどうすればよいでしょうか？

　以下自分がやってみたこと。

分散を計算するときの桁落ち対策

分散 = 二乗平均 - 期待値の二乗

という式を使って分散を計算すると、右辺の2つの項の値が非常に近い場合、桁落ちが発生する。

たとえば以下のようなデータを考える。

x₁ = 10¹⁰
x₂ = 10¹⁰ + 1

x₃ = 10¹⁰ + 2

x₄ = 10¹⁰ + 3

x₅ = 10¹⁰ + 4

この場合、二乗平均も期待値の二乗も10²⁰程度である。

両者の差は2だが、倍精度浮動小数点の有効桁数は15桁程度なので、この差の情報は失われてしまう。

分散は平行移動しても変わらないので、平均が0になるように中心化を行う。

x₁ = -2
x₂ = -1

x₃ = 0

x₄ = 1

x₅ = 2

すると、二乗平均は2、期待値の二乗は0となり、差の情報は失われない。

以下のプログラムを実行してみると、桁落ちの様子を確認できる。

#include <iostream>
#include <vector>
#include <iomanip>

using namespace std;

// データを中心化する
vector<double> centerize(const vector<double> &v) {
  int n = v.size();
  double avg = 0.0;
  for (auto &x : v)
    avg += x;
  avg /= n;
  vector<double> w(n);
  for (int i = 0; i < n; i++)
    w[i] = v[i] - avg;
  return w;
}

// 分散を計算する
double calc(const vector<double> &v) {
  int n = v.size();
  double avg_sq = 0.0;
  double avg = 0.0;
  for (auto &x : v) {
    avg += x;
    avg_sq += x * x;
  }
  avg_sq /= n;
  avg /= n;

  return avg_sq - avg * avg;
}

// データ生成
vector<double> gen() {
  vector<double> v;
  for (int i = 0; i < 5; i++)
    v.push_back(1e10 + i);
  return v;
}

int main(int argc, char *argv[])
{
  vector<double> v = gen();
  cout << fixed << setprecision(15);
  cout << calc(v) << endl;
  cout << calc(centerize(v)) << endl;
  return 0;
}

以下実行結果。

$ ./main       
0.000000000000000
2.000000000000000

SRM 712 Div1 600 AverageVarianceSubtree

問題
ノードに重みがつけられた木が与えられる。
木のすべての部分木について重みの分散を計算し、この分散の平均値を求めよ。

解法
DFSしてノード上でDPする頻出テクニックを使う問題だが、分散を計算するために一工夫必要。すべての部分木の分散を効率よく計算するためには（ルート, 部分木のサイズ）ごとに"重みの二乗和"と"重みの和の二乗"を保持すればよい。

例として、ノード(a, b)からなる木とノード(c, d)からなる木をマージする作業を考えてみる。
木(a, b)の二乗和と木(c, d)の二乗和がわかっていたとすると、マージした木の二乗和はそのままこの二つを足せばいいだけなので簡単。

問題は、和の二乗の方で、
(a + b + c + d)^2 = a^2 + b^2 + c^2 + d^2 + 2ab + 2ac + 2ad + 2bc + 2bd + 2cd
(a + b)^2 = a^2 + 2ab + b^2
(c + d)^2 = c^2 + 2cd + d^2
となるので、単純に足すわけにはいかない。

和の二乗 = 二乗の和 + 2 * 異なる項の積和

となっているので、異なる項の積和をうまく部分問題から計算できればよいことが分かる。

木(a, b)の異なる項の積和 = ab
木(c, d)の異なる項の積和 = cd

木(a, b, c, d)の異なる項の積和 = ab + ac + ad + bc + bd + cd = ab + cd + (a + b)(c + d)

となるので、部分問題の異なる項の積和と部分問題の和を使えば計算できることが分かる。

以上をふまえて、

• sum[v][i] = ノードvを根とするサイズiの木の重み和
• sum_sq[v][i] = ノードvを根とするサイズiの木の重み二乗和
• sum_prd[v][i] = ノードvを根とするサイズiの木の重みの異なる項の積和

を葉から根方向にDPさせればよい。木をマージするときは、マージする相手方のサイズの分だけ自身が繰り返し現れるので、その分を掛けるのを忘れないようにする。

あとこの問題は数値計算の精度が厳しいらしく、計算誤差を小さくするため以下の工夫が必要。

• あらかじめ重みを中心化しておく。
• long doubleを使う。

ソースコード

using namespace std;

#define ALL(x) (x).begin(), (x).end()
#define EACH(itr,c) for(__typeof((c).begin()) itr=(c).begin(); itr!=(c).end(); itr++)  
#define FOR(i,b,e) for (int i=(int)(b); i<(int)(e); i++)
#define MP(x,y) make_pair(x,y)
#define REP(i,n) for(int i=0; i<(int)(n); i++)

int n;
long long sz[55][55];
double long sum[55][55];
double long sum_sq[55][55];
double long sum_prd[55][55];
double long w[55];
vector<int> child[55];

void dfs(int v) {
  REP (i, n+1) {
    sz[v][i] = 0;
    sum[v][i] = sum_sq[v][i] = sum_prd[v][i] = 0.0;
  }

  sz[v][1] = 1;
  sum[v][1] = w[v];
  sum_sq[v][1] = w[v] * w[v];

  for (auto &c : child[v]) {
    dfs(c);
    for (int i = n; i > 1; i--) {
      for (int j = 1; j < i; j++) {
        int k = i - j;
        sz[v][i] += sz[v][j] * sz[c][k];
        sum[v][i] += sum[v][j] * sz[c][k] + sum[c][k] * sz[v][j];
        sum_sq[v][i] += sum_sq[v][j] * sz[c][k] + sum_sq[c][k] * sz[v][j];
        sum_prd[v][i] += sum_prd[v][j] * sz[c][k] + sum_prd[c][k] * sz[v][j] + sum[v][j] * sum[c][k];
      }
    }
  }
}

class AverageVarianceSubtree {
  public:
  double average(vector<int> p, vector<int> weight) {
    n = weight.size();
    double long avg = 0.0;
    REP (i, n) avg += weight[i];
    avg /= n;
    REP (i, n) w[i] = weight[i] - avg;
    REP (i, n) child[i].clear();
    REP (i, p.size()) child[p[i]].push_back(i+1);

    dfs(0);

    long long tot = 0;
    long double ret = 0.0;

    REP (v, n) {
      for (int i = 1; i <= n; i++) {
        tot += sz[v][i];
        ret += sum_sq[v][i] / i - (sum_sq[v][i] + 2 * sum_prd[v][i]) / i / i;
      }
    }
    
    return ret / tot;
  }
};