めもめも

このブログに記載の内容は個人の見解であり、必ずしも所属組織の立場、戦略、意見を代表するものではありません。

数学を学ぶ動機に関するポエム

最近ある技術書の筆者の方が書籍紹介のブログ記事の中で「俺は数学が嫌いだ」と言い放たれていて、ひどくショックを受けたので、心の傷を癒やすために、なぜ僕はこんなにも数学が気になって、数学を勉強したくなるか、頭の中を整理してみました。

ちなみに僕がどのぐらい数学が気になるかというと、もし宝くじで5兆円ぐらい手に入ったら、個人で数学の研究機関を設立して、優秀な数学者を囲い込んで、僕が興味のあることだけを研究させて、毎日、僕のためだけに講義をさせるとか、そんなことをやってみたいぐらい、数学が気になります。

なお、以下の内容は、時系列っぽく書かれていますが、実際に僕がこのような順番で数学に興味を持って理解していったわけではありません。ただ、振り返ってみると、こういった要素が折り重なって、数学に対する興味が構成されていると気づいたというような内容です。

第一段階(みんなが同じルールに従っていることを確認したい)

みなさんは、子供のころなんとなく「物の数を数える」ということを教わってできるようになったと思います。人間には、「数学っぽい」ことをする能力が生まれながらにしてあるようです。

ですが、自分の「数学っぽい」やり方と、他の人の「数学っぽい」やり方がまったく同じ考え方、まったく同じルールに従っていると自信を持っていえるでしょうか? もしかしたら、自分の物の数え方と、他の人の物の数え方には、何か違うルールがあるかもしれません。そもそも、自分のやり方に本当に一貫性があるのかもあやしいです。

そこで、自分がやっている「数学っぽい」ことを論理的な言葉で明文化して、みんなが同じ考え方、同じルールで処理していることを確認したくなります。

たとえば、僕は次のような計算を行います。

3\times -2 = -6
-2\times 5 = -10

ここでは、正の数と負の数をかけると負の数になるというルールに従います。

「なぜそうなるのか?」というのは別にして、まずは、万人がみなこのルールに従っているということを確認する必要があります。そのためには、「正の数」と「負の数」を定義して、さらに、「正の数と負の数の積」を計算する手順を定義する必要があります。

まあ、実際には自分で定義する必要はなくて、先人の数学者が定義してくれたお言葉が書かれた「数学書」があります。そこで、それらを何冊か読んで、すべてにちゃんと同じ手順が書かれていることを知ります。これで僕は「少なくともこういった数学書を書いている/読んでいる数学者の仲間の中では、みんな僕と同じルールに従っているんだ」と安心することができます。あーよかった。

第二段階(自分のやり方が正しいことを納得させたい)

数学のルールをいくつか覚えると、このルールに従って、自分一人でいろんな計算ができるようになります。何か面白い事実が証明できるかも知れません。すると、それを他人に説明したくなります。この時、「他の人が自分と同じルールに従って計算している」という前提が超重要になります。この前提がくずれると、いくら一生懸命説明しても、相手は、謎の理屈で論破してくるかも知れません。

「正の数をかけるっていうのは、それだけ足し合わせるということだろ。だったら正の数をかけて小さくなるのはおかしいだろ。 -2\times 5-2 より大きい数でないとおかしい!」

こういう相手とは議論が成り立ちません。数学というのは、基本的なルールを共有していることを前提にして、そのルールに従って一緒に議論を進めて、自分と相手が同じ結論に達することを確認しあうという、一種のゲームなのです。

まぁもちろん、ながーい計算やら証明をする際は、細かいルールは自明のものと思って、説明を端折ることはあります。でも、相手に「そこはどういうルールで話が進んだのか分からん」と指摘された際に、端折った部分をちゃんと一から説明して、相手を納得させることができればOKです。同じルールを共有することで、こういった知的なゲームを楽しむことができるようになります。

えー。

「お前はほんとにそんなゲームをやるのか。妄想じゃねえの?」

という方へ。

僕は大学生のころ「数学演技」という講座をとっていました(「演習」ではなくて「演技」というあたりがカッチョいいです)。教授から証明問題を与えられて、黒板の前に立って、その証明を進めていきます。時々、教授が「そこの式変形が成立する理由がわからん。もっとくわしく」と突っ込むので、心の底で「きたきたきたきたきたきたきたきたーーーー」と思いながら、端折った部分を根掘り葉掘り説明します。どこまで詳しく説明するかは、教授の顔色を見て考えます。最終的に僕の証明に教授が納得すれば、僕の勝利です。このゲームのゴールは、数学的に何かを証明することではありません。「教授を納得させること」がゴールなのです。

まあ、よく考えると、数学の世界って、そもそもそんなものかも知れません。ある数学者が何かを「証明した!」と言って論文を発表します。世の中の数学者がそれを読んで、「なるほど。これは正しい」と納得すれば、その証明は正しいものとして受け入れられます。「何かが数学的に正しいかどうか」というのは、究極的には、世の中の数学者の大部分が「それは正しい」と納得するかどうかで判断するしかありません。もちろん、しつこく言いますが、世の中の数学者が基本的なルールをきちんと共有しているからこそ、それでうまくいくのです。

ただ過去の歴史を振り返ると、ある時代まではほぼすべての数学者が「これが成り立つのは当たり前だよね」と思っていたことが、ある数学者によって、実は当たり前ではないことが指摘されて、これまで正しいとされていたことが、正しくなくなることもあり得ます。あることが「数学的に正しい」かどうかは、その時代の数学者たちがどのような前提を共有するかで変わり得るということかも知れません。

大学一年生が解析学を習うと必ずでてくる「ε-δ論法」ありますよね。直感的に自明な極限の計算をするのに、なんで、あんな七面倒な議論が必要なのでしょう? それは、「共通言語」が必要だからです。「直感的に成立する」と主張するのは自由ですが、他人が同じ直感を持っていると無条件に仮定するわけにはいきません。直感的に当たり前なことを万人が共有できる論理的な形で表現するのは、意外と難しいのです。「極限」という概念をなんとか論理的に表現する苦肉の策が「ε-δ論法」なのです。数学演技で教授を納得させるには、自分と教授の間で合意のとれた「ε-δ論法」を使うしかないのです。

第三段階(自分のやり方に矛盾がないか確かめたい)

数学の不思議なところは、ある計算をするのに複数のやり方があって、どの方法でもちゃんと同じ結果に行き着くということです。ぶっちゃけ、不思議じゃないですか???? 数学にはいろんな「ルール」がありますが、ルールAとルールBで答えがちがうようなことはないのです。ここはもう納得の問題ですが、ある人は、こんなふうに考えて納得します。

「この世界には、『矛盾のない数学的真理』というものがあって、我々は知性を駆使してその真理を明らかにしていっているのだ。もし矛盾があれば、それは我々の知性が真理を正しく捉えていないのだ。俺は真理を信じるのだ」

まぁ。納得するための方法としてはあり得るかもしれませんが・・・・。僕はこれでは納得できませんでした。数学のルールというのは、あくまで人間が作ったもので、人間の外に数学的真理なるものがあるのかどうか、僕にはわかりません。存在を否定するわけではありませんが、あるかどうかわからないものを前提に数学の議論を進めることは、とても気持ち悪くて耐えられません。兎に角、暗黙のルールを徹底的に明文化して、整理して、「最低限これだけのルールから出発すれば、数学のルールはすべて、ここから導くことができる」という最小構成のルールセットを見つければ、もしかしたら、そこから数学に矛盾がないことを証明できるかも知れません。

で、実は世の中には、まさにそういう研究をしている数学者がいて、「数学基礎論」とか呼ばれていたりします。ゲーデル先生は、その昔、数学の「形式表現」という仕事に取り組みました。

これは、数学のすべてのルールを「記号変形のルール」として表現しようというものです。数学の計算をする時、一般人は、その「意味」を考えながら計算を進めます。「2+3」を計算する時は、何か物が2個と3個あるようすを想像して、全部で5個になるから「2+3=5」なんだと納得します。でも、コンピューターが計算する時は、そんな事はおそらく考えてはいません。決められたルールに従って、ビットの並びを操作しているだけです。

このようにコンピューターが自動計算できるように、機械的な記号変形のルールを定めて、それで世の中のすべての数学を表してしまえ、という壮大な構想です。

で、ゲーデル先生はやりました。詳しくは書きませんが、10個程度の基本ルールを定めて、そのルールにしたがって記号変形をすすめれば、世の中の数学のほぼ大半の理論が再現できることを示したのです。すごいですよね。むっちゃ興奮しませんか? 数学的真理なるものが人間の外にあるかどうかはもはや関係ありません。この10個程度のルールが数学的真理なのです。

この基本ルールを元にして、数学に矛盾がないかどうかを探求しようというのが、数学基礎論です。現在では、ある程度の結論は出ていて、その内容は僕もある程度知っているのですが、その結論にいたる道筋はまだ完璧に理解できていません。死ぬまでに一度、これを完璧に理解して、「おれは本当の意味で数学を理解した!!!」とか叫んで走り回ってみたいものです。

第四段階(数学が表現できる世界の限界を知りたい)

数学基礎論の立場にたてば、10個程度の基本ルールにしたがってひたすら記号変形を進めれば、あらゆる数学的真理、つまり、「証明可能な論理式」がすべて得られます。とはいえ、いわゆる組合せ爆発というやつで、何も考えずにランダムに変形しても面白い結果にはたどり着きません。面白い結果にたどりつくには、記号に対して「意味」を与えて、論理式が示す内容を「解釈」する必要があります。解釈が得られれば、自分が持つ「数学的直感」に従って、こんな事も成り立つかもしれない、という想像ができます。そして、それが本当になりたつかどうか、ランダムな記号変形ではなく、人としての意思を持った式変形を進めることができます。人間のもつ「直感」の世界はおそろしく広いです。この膨大な直感の世界のいったい、どこからどこまでの範囲を数学は(一定のルールに従った記号変形という操作の中で)表現して、証明することができるのでしょうか?

ある言語学者は、「人間は言語を使って思考するから、人間の思考は言語の構造によって限界付けられる」というような事をいいます。もしかしたら、「いや俺/私は言語をつかわずに思考している」と言う人もいるかも知れません。ただ、すくなくともその思考を表現して他人と共有するには言語が必要です。そういう意味で、少なくとも他人と共有可能な思考は、言語の構造で限界づけられるようです。

数学における「直感」も同じかも知れません。人間は生まれながらにして、「数学っぽい」ことを直感的に理解する能力があります。でも、その直感を明文化して他人と理解を共有するには、正しい数学のお作法に則って表現する必要があります。この意味で、人間の「数学的思考力」は、数学の構造によって限界づけられると思っています。数学が表現できること、証明できることの限界を探るということは、人間の「数学的思考力」の限界に対する挑戦なのかも知れません。いや、これ、むっちゃ気になりますよね。人間の思考力の限界ですよ。だからこそ、僕は「厳密な数学」が大好きなのです。厳密な数学の限界を知ることで、人間の思考力の限界を知りたいのです。あいまいな直感的な議論では、真の意味における「思考力の限界」は探れないのです。

第五段階(今まで気づいていなかった世界を見たい)

上の議論は「数学的直感」が先にあって、それをお作法に則った「厳密な数学」でどこまで表現できるか、という観点でした。ですが、世の中の数学者は、そもそも僕の直感には無かったような斬新な事実を「厳密な数学」の言葉で表現してきます。すごいです。彼らも人間ですから、機械的に式変形しているだけではないはずです。彼らの脳内には、なんらかの直感的な世界・構造・仕組みの理解があって、それを表現しているはずです。ただそれは、高度に抽象化されすぎていて、もはや厳密な数学の言葉でしか表現できないのかも知れません。あるいは、機械的な式変形の可能性を探る中で、何か新しい直感的事実を見つけ出したのかも知れません。どっちでもいいです。僕もなんとかして、同じ直感的理解を共有したいです。彼らが見ている世界を僕も見てみたいです。

これって絵画の世界と似ているかも知れませんね。目で見た世界をそのまま忠実に写し取る絵画の世界から、新しい表現技法が生まれて、いままでなかった抽象画の世界ができました。色を並べるという操作で、いったいどれだけの事が表現できるのか。まだ誰も気づいていなかった、新たな表現の世界はないのか、絵画は人の心を映したものであると同時に、ときには、人の心が捉えられる世界を広げるという役割を持ちます。

詩や小説の世界も同じですよね。有限の個数の文字を並べることで、いったいどれだけの世界が表現できるのでしょうか? 言葉による表現の限界への挑戦ですよね。

そういう意味で、「厳密な数学」が表現できる世界の広さは、僕の素朴な直感の世界をはるかに超えています。まだ見えていない世界をもっともっと見たいです。だから数学は面白いのです。

Jupyterノートブック上でJavaScriptからカーネルのコードを実行する方法

qiita.com

これは、jupyter notebook Advent Calendar 2016の10日目の記事です。

何の話かというと

IPython.displayモジュールのHTML関数を使うと、次のように、Jupyterノートブック上でJavaScriptを実行することができます。

from IPython.display import HTML

javascript = '''
<script type="text/javascript">
    alert("HOGE")
</script>
'''

HTML(javascript) # ポップアップウィンドウを表示

この時さらに、JavaScriptからノートブックを実行中のカーネルを呼び出して、任意のコードを実行することが可能です。また、JavaScript側でコードの実行結果を受け取ることもできます。具体的には、次のようになります。

import numpy as np
from IPython.display import HTML

javascript = '''
<script type="text/javascript">
var kernel = IPython.notebook.kernel;

var callback = function(output) {
  var res = output.content.data['text/plain'];
  alert(res);
};

var x = 2.0;
var command = 'np.sin(' + x + ')';
kernel.execute(command, {'iopub': {"output": callback}}, {silent:false});
</script>
'''

HTML(javascript)

この例では、Pythonカーネル側で np.sin(2.0) を計算して、その結果をJavaScriptに返しています。カーネルの呼び出しは非同期に行われるので、カーネル側の計算が終わったタイミングでコールバック関数(この例では callback)が呼び出されます。JavaScriptからAjaxでREST APIを呼ぶ感覚で、カーネル側のコードを実行すると思えばよいでしょう。

やってみた

というわけで、これを応用して、Jupyterノートブックで遊べるオセロゲームを作りました。オセロの盤面の処理(コマを置いてひっくり返すとか)とコンピュータ側の思考ルーチンをPythonで実装して、JavaScriptのUIから呼び出すようにしています。UIの実装は、enchant.jsを使っています。

github.com

実行画面はこんな感じ。囲碁のパーツで作ったので超違和感がありますがwww。あと、ゲームの終了判定や「パス」ボタンの実装は、面倒だったので省略しています。

うまくいかない点

ゲーム画面が描画された後、画面を上下にスクロールするとクリック位置の判定がおかしくなります。(クリック位置が固定されたままで、一緒にスクロールしていない。)

enchant.jsの実行ループを開始すると、カーソルキーが無効になります。

むむぅ。誰が解決策がわかったら教えてください。

おまけ

コンピュータ側の思考ルーチンはMini-Max法を使っています。具体的には、次のクラスのget_values()メソッドで盤面の評価値(人間側がどの程度優勢か)を計算して、これをなるべく小さくするように打ってきます。

class SimpleMiniMax:
    def get_values(self, boards):
        result = []
        for board in boards:
            score = 0.0
            # The value is defined from the player=1's point of view.
            for c in sum(board, []):
                if c == 1: score += 1
                if c == -1: score -= 1
            result.append([score])
        return np.array(result)

この実装では、「人間側のコマの数-コンピュータ側のコマの数」を評価値にしているので、人間側のコマをなるべく減らす(コンピュータ側のコマをなるべく増やす)という単純なアルゴリズムになっています。最近はやりのディープなんとかで、評価値の計算をなんとかすると、なんとかなあれができるかも知れません。がんばってください。

Using Cloud Dataflow to run parallel predictions with your TensorFlow model

Suppose that you've finished training your prediction model with TensorFlow, yeay!

Now you have to make predictions with the trained model for tens of thousands of data. How will you do it?

I will show you one of possible choices. You can use Cloud Dataflow for general parallel batch processing and it's not hard to apply machine learning models in this framework.

What is Cloud Dataflow?

Cloud Dataflow is a runtime environment running on Google Cloud Platform for parallel data processing. Traditionally you may have written custom MapReduce codes with Hadoop, but it often becomes too complicated to optimise when you have to combine multiple MapReduce jobs. Instead, Cloud Dataflow builds optimised MapReduce jobs from a dataflow description code written with the Apache Beam SDK.

Here's a code snippet for the wordcount example.

import apache_beam as beam

class WordExtractingDoFn(df.DoFn):
  def process(self, context):
    text_line = context.element.strip()
    words = re.findall(r'[A-Za-z\']+', text_line)
    return words

p = beam.Pipeline()
lines = p | 'Step1. Read lines' >> beam.io.Read(beam.io.TextFileSource(input_file))
words = (lines | 'Step2. Split into words' >> beam.ParDo(WordExtractingDoFn()))
counts = (words | 'Step3. Assign one to each word' >> beam.Map(lambda x: (x, 1))
                | 'Step4. Group by word' >> beam.GroupByKey()
                | 'Step5. Do count' >> beam.Map(lambda (word, ones): (word, sum(ones)))
         )
output = counts | 'Step6. Format' >> beam.Map(lambda (word, c): '%s: %s' % (word, c))
output | 'Step7. Output results' >> beam.io.Write(beam.io.textio.WriteToText(output_file))
p.run()

Step2&3 corresponds to Map jobs, step4 corresponds to the Shuffle phase, and step5&6 corresponds to Reduce jobs. When you run this code on your SDK environment (typically on the Cloud Shell), it throws a job to the runtime environment and corresponding MapReduce jobs are automatically executed.

How to write codes for Cloud Dataflow jobs?

In the wordcount example, the WordExtractingDoFn class provides a custom Map job to split a single line into a list of words. In the same manner, you can write a class to make a prediction with a trained TensorFlow model for a single data. The jobs are executed in multiple containers on the runtime environment, and you can add additional modules and binaries in the container if necessary. In this case, I will add the TensorFlow module in the runtime.

In the following example, I use three files to run predictions for the MNIST dataset with Cloud Dataflow.

・setup.py : A boiler plate script to customise the runtime. I use this to install the TensorFlow module.
・run.py : A thin wrapper to kick a job.
・module/predict.py : The main code to build and run the dataflow pipeline.

In addition, a text file containing the MNIST dataset and a TensorFlow's model checkpoint file are uploaded in Cloud Storage. (The bucket name dataflow99 can be arbitrary.)

・MNIST dataset: gs://dataflow99/input/images.txt
・Checkpoint file: gs://dataflow99/input/cnn_session-20000

The MNIST dataset file is created with the following code. Each line contains a single image and the data number is added at the top of the line.

from tensorflow.examples.tutorials.mnist import input_data

mnist = input_data.read_data_sets("/tmp/data/", one_hot=True)
with open('images.txt', 'w') as file:
    for num, image in enumerate(mnist.test.images):
        file.write('%d,%s' % (num, ','.join(map(str,image.tolist()))))
        file.write('\n')

Now let's go through the actual codes.

setup.py

from distutils.command.build import build as _build
import setuptools
import subprocess

class build(_build):  # pylint: disable=invalid-name
  sub_commands = _build.sub_commands + [('CustomCommands', None)]

class CustomCommands(setuptools.Command):
  def initialize_options(self):
    pass

  def finalize_options(self):
    pass

  def RunCustomCommand(self, command_list):
    print 'Running command: %s' % command_list
    p = subprocess.Popen(
        command_list,
        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    # Can use communicate(input='y\n'.encode()) if the command run requires
    # some confirmation.
    stdout_data, _ = p.communicate()
    print 'Command output: %s' % stdout_data
    if p.returncode != 0:
      raise RuntimeError(
          'Command %s failed: exit code: %s' % (command_list, p.returncode))

  def run(self):
    for command in CUSTOM_COMMANDS:
      self.RunCustomCommand(command)


CUSTOM_COMMANDS = [['pip', 'install', 
  'https://storage.googleapis.com/tensorflow/linux/cpu/tensorflow-0.10.0-cp27-none-linux_x86_64.whl']]
REQUIRED_PACKAGES = []

setuptools.setup(
    name='tensorflow-module',
    version='0.0.1',
    description='TensorFlow model prediction package.',
    install_requires=REQUIRED_PACKAGES,
    packages=setuptools.find_packages(),
    cmdclass={'build': build, 'CustomCommands': CustomCommands}
)

This is almost a boiler plate. The important parts are CUSTOM_COMMANDS and REQUIRED_PACKAGES. In this example, the TensorFlow module is explicitly installed with the pip command.

run.py

import logging
from modules import predict
if __name__ == '__main__':
  logging.getLogger().setLevel(logging.INFO)
  predict.run()

This is a thin wrapper to run the module 'modules/predict.py'.

modules/predict.py

import tensorflow as tf
import numpy as np
import apache_beam as beam
import argparse
import logging

def singleton(cls):
  instances = {}
  def getinstance():
      if cls not in instances:
          instances[cls] = cls()
      return instances[cls]
  return getinstance

@singleton
class Model():
  def __init__(self):
    num_filters1 = 32
    x = tf.placeholder(tf.float32, [None, 784])
    x_image = tf.reshape(x, [-1,28,28,1])
    
    W_conv1 = tf.Variable(tf.truncated_normal([5,5,1,num_filters1],
                                              stddev=0.1))
    h_conv1 = tf.nn.conv2d(x_image, W_conv1,
                           strides=[1,1,1,1], padding='SAME')
    b_conv1 = tf.Variable(tf.constant(0.1, shape=[num_filters1]))
    h_conv1_cutoff = tf.nn.relu(h_conv1 + b_conv1)
    h_pool1 =tf.nn.max_pool(h_conv1_cutoff, ksize=[1,2,2,1],
                            strides=[1,2,2,1], padding='SAME')
    
    num_filters2 = 64
    W_conv2 = tf.Variable(
                tf.truncated_normal([5,5,num_filters1,num_filters2],
                                    stddev=0.1))
    h_conv2 = tf.nn.conv2d(h_pool1, W_conv2,
                           strides=[1,1,1,1], padding='SAME')
    b_conv2 = tf.Variable(tf.constant(0.1, shape=[num_filters2]))
    h_conv2_cutoff = tf.nn.relu(h_conv2 + b_conv2)
    h_pool2 =tf.nn.max_pool(h_conv2_cutoff, ksize=[1,2,2,1],
                            strides=[1,2,2,1], padding='SAME')
    h_pool2_flat = tf.reshape(h_pool2, [-1, 7*7*num_filters2])
    
    num_units1 = 7*7*num_filters2
    num_units2 = 1024
    w2 = tf.Variable(tf.truncated_normal([num_units1, num_units2]))
    b2 = tf.Variable(tf.constant(0.1, shape=[num_units2]))
    hidden2 = tf.nn.relu(tf.matmul(h_pool2_flat, w2) + b2)
    
    keep_prob = tf.placeholder(tf.float32)
    hidden2_drop = tf.nn.dropout(hidden2, keep_prob)
    
    w0 = tf.Variable(tf.zeros([num_units2, 10]))
    b0 = tf.Variable(tf.zeros([10]))
    p = tf.nn.softmax(tf.matmul(hidden2_drop, w0) + b0)
    
    sess = tf.InteractiveSession()
    sess.run(tf.initialize_all_variables())
    saver = tf.train.Saver()
    saver.restore(sess, 'gs://dataflow99/input/cnn_session-20000')

    self.sess, self.p, self.x, self.keep_prob = sess, p, x, keep_prob


class PredictDoFn(beam.DoFn):
  def process(self, context):
    model = Model()
    image = context.element.split(',')
    datanum = image.pop(0)
    pred = model.sess.run(model.p,
                          feed_dict={model.x:[image], model.keep_prob:1.0})[0]
    result = '%s: %s' % (datanum, ','.join(map(str, pred.tolist())))
    return [result]


def run(argv=None):
  parser = argparse.ArgumentParser()
  parser.add_argument('--input', dest='input', required=True,
                      help='Input file to process.')
  parser.add_argument('--output', dest='output', required=True,
                      help='Output file to write results to.')
  known_args, pipeline_args = parser.parse_known_args(argv)

  p = beam.Pipeline(argv=pipeline_args)
  images = p | 'ReadFromText' >> beam.io.Read(beam.io.TextFileSource(known_args.input))
  predictions = (images | 'Prediction' >> beam.ParDo(PredictDoFn()))
  predictions | 'WriteToText' >> beam.io.Write(beam.io.textio.WriteToText(known_args.output))
  logging.getLogger().setLevel(logging.INFO)
  result = p.run()

This executes the actual prediction jobs. The Model class provides the double layer CNN model explained in the TensorFLow Tutorials. The pretrained checkpoint is restored from gs://dataflow99/input/cnn_session-20000.

Note that the Model class is implemented as a singleton. Since the PredictDoFn(beam.DoFn) class could be instantiated multiple times in the same thread, you need to avoid creating multiple TensorFlow sessions in this way.

Run it!

The Apache Beam SDK is preinstalled in the Cloud Shell. So you can directly run the script on the Cloud Shell. The following runs the codes locally on the Cloud Shell to test them. The results are written in /tmp/OUTPUT-00000-of-00001.

$ BUCKET=gs://dataflow99
$ PROJECT=<your project ID>
$ python run.py --input $BUCKET/input/images.txt --output /tmp/OUTPUT

The following runs the job on the Cloud Platform. Runtime environment is automatically deployed on Google Compute Engine and destroyed after the execution.

$ python run.py --runner BlockingDataflowPipelineRunner \
                --project=$PROJECT \
                --staging_location $BUCKET/staging \
                --temp_location $BUCKET/temp \
                --setup_file ./setup.py \
                --job_name $PROJECT-prediction \
                --input $BUCKET/input/images.txt \
                --output $BUCKET/output/predictions

The results are written in multiple files on Cloud Storage gs://dataflow99/output/predictions-0000[0...9]-of-00009. Here are a few examples of the predictions.

7831: 4.27848138862e-11,4.68908603352e-13,2.48020860454e-05,0.999975204468,7.18904258557e-17,6.26192708797e-11,2.0732427408e-20,9.44890032883e-09,1.42062542285e-11,3.0826359243e-09
7832: 2.86281824913e-08,0.999999403954,1.5229447925e-08,2.67739736631e-12,5.05345951751e-07,2.36653296959e-09,1.38168243513e-09,7.38670680178e-09,3.35720322653e-08,1.15596296424e-10
7833: 1.49982966951e-19,2.73763130221e-12,1.07422761838e-15,2.25128758385e-19,1.0,4.67769159385e-15,3.64918932859e-17,2.55687953449e-11,1.83131521693e-14,5.16055809197e-11
7834: 4.4291605619e-20,4.20704703713e-13,4.15420612127e-16,5.78066899277e-17,1.0,2.73779564693e-14,5.10855258428e-16,8.55861992388e-10,4.34696029172e-11,4.51999015993e-09

Disclaimer: All code snippets are released under Apache 2.0 License. This is not an official Google product.