今回はBatch&StreamフレームワークであるApacheBeamを動かすための環境を構築してゆきます。
ようやく、アプリエンジニアっぽいことができそうです。
目次
前提記事
今回の範囲
- ApacheBeamの概要
- Python版ApacheBeam環境の構築
- 動作確認
ApacheBeam概要
Beamとは冒頭で書いた通り、BatchとStreamの造語で、名前の通りリアルタイムな大規模データを処理するために開発されたフレームワークです。
また、分散処理が簡単に書けるような文法が用意されています。
GoogleDataFlowで計算量を算出し、自動スケールするBeam環境を構築することができるなど、徐々に話題になりつつあるフレームワークです。
今後、機械学習と組み合わせたり、大規模なデータを扱うプログラムの開発に利用してゆきたいと考えております。
Python版ApacheBeam環境の構築
公式ドキュメントを参考に構築してゆきます。 (AmazonLinuxでいけるのかドキドキです)
https://beam.apache.org/get-started/quickstart-py/
> The Beam SDK requires Python 2 users to use Python 2.7 and Python 3 users to use Python 3.5 or higher
> Install pip, Python’s package manager. Check that you have version 7.0.0 or newer by running:
とありますね(執筆時)
Cloud9は標準で以下の環境が入っているので、クリアしています。
$ python -V
Python 3.6.8
$ pip -V
pip 9.0.3 from /usr/lib/python2.7/dist-packages (python 2.7)
新たにプロジェクトディレクトリを作成し、venvを作成し、Activate
ec2-user:~/environment/Python_Project $ mkdir ApacheBeamTest
ec2-user:~/environment/Python_Project $ cd ApacheBeamTest/
ec2-user:~/environment/Python_Project/ApacheBeamTest $ ls
ec2-user:~/environment/Python_Project/ApacheBeamTest $ python -m venv beamenv
ec2-user:~/environment/Python_Project/ApacheBeamTest $ source beamenv/bin/activate
(beamenv) ec2-user:~/environment/Python_Project/ApacheBeamTest $
ApacheBeamをインストール
$ pip install apache-beam
※gcsやテスト、ドキュメント生成のライブラリも用意されているようですね。一旦今回は無視してファイルの入出力を実施します。

以下のgitリポジトリより単語数を算出するプログラムをコピペして、エディタでファイルを作成します。
https://github.com/apache/beam/tree/master/sdks/python/apache_beam/examples
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""A word-counting workflow."""
from __future__ import absolute_import
import argparse
import logging
import re
from past.builtins import unicode
import apache_beam as beam
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from apache_beam.metrics import Metrics
from apache_beam.metrics.metric import MetricsFilter
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
class WordExtractingDoFn(beam.DoFn):
"""Parse each line of input text into words."""
def __init__(self):
# TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.
# super(WordExtractingDoFn, self).__init__()
beam.DoFn.__init__(self)
self.words_counter = Metrics.counter(self.__class__, 'words')
self.word_lengths_counter = Metrics.counter(self.__class__, 'word_lengths')
self.word_lengths_dist = Metrics.distribution(
self.__class__, 'word_len_dist')
self.empty_line_counter = Metrics.counter(self.__class__, 'empty_lines')
def process(self, element):
"""Returns an iterator over the words of this element.
The element is a line of text. If the line is blank, note that, too.
Args:
element: the element being processed
Returns:
The processed element.
"""
text_line = element.strip()
if not text_line:
self.empty_line_counter.inc(1)
words = re.findall(r'[\w\']+', text_line, re.UNICODE)
for w in words:
self.words_counter.inc()
self.word_lengths_counter.inc(len(w))
self.word_lengths_dist.update(len(w))
return words
def run(argv=None, save_main_session=True):
"""Main entry point; defines and runs the wordcount pipeline."""
parser = argparse.ArgumentParser()
parser.add_argument('--input',
dest='input',
default='gs://dataflow-samples/shakespeare/kinglear.txt',
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)
# We use the save_main_session option because one or more DoFn's in this
# workflow rely on global context (e.g., a module imported at module level).
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
p = beam.Pipeline(options=pipeline_options)
# Read the text file[pattern] into a PCollection.
lines = p | 'read' >> ReadFromText(known_args.input)
# Count the occurrences of each word.
def count_ones(word_ones):
(word, ones) = word_ones
return (word, sum(ones))
counts = (lines
| 'split' >> (beam.ParDo(WordExtractingDoFn())
.with_output_types(unicode))
| 'pair_with_one' >> beam.Map(lambda x: (x, 1))
| 'group' >> beam.GroupByKey()
| 'count' >> beam.Map(count_ones))
# Format the counts into a PCollection of strings.
def format_result(word_count):
(word, count) = word_count
return '%s: %d' % (word, count)
output = counts | 'format' >> beam.Map(format_result)
# Write the output using a "Write" transform that has side effects.
# pylint: disable=expression-not-assigned
output | 'write' >> WriteToText(known_args.output)
result = p.run()
result.wait_until_finish()
# Do not query metrics when creating a template which doesn't run
if (not hasattr(result, 'has_job') # direct runner
or result.has_job): # not just a template creation
empty_lines_filter = MetricsFilter().with_name('empty_lines')
query_result = result.metrics().query(empty_lines_filter)
if query_result['counters']:
empty_lines_counter = query_result['counters'][0]
logging.info('number of empty lines: %d', empty_lines_counter.result)
word_lengths_filter = MetricsFilter().with_name('word_len_dist')
query_result = result.metrics().query(word_lengths_filter)
if query_result['distributions']:
word_lengths_dist = query_result['distributions'][0]
logging.info('average word length: %d', word_lengths_dist.result.mean)
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run()
こんな感じ

適当にテキストファイルを作成します。
hello, world!!!!

実行してみます。
$ python -m apache_beam.examples.wordcount --input hw.txt --output out.txt
ばしばしログが出て、結果ファイルが出ます。

簡単に実行できましたね。
おわりに
今後はこのフレームワークを活用しながらビッグデータ解析などやってみようかなぁと思います。
あまり日本語情報がないので、それぞれのプログラムの説明なども出きればと思っています。