概要

BigQueryに保存されたtweetを形態素解析したい。BigQuery単体では形態素解析出来ないし、mecabの新語辞書を使ったりも出来ないのでdataflowを利用することで形態素解析を行う。
twitterでは、常に最新の言葉が使われる。そのためmecabの辞書は neologd を利用して最新の状態の言葉で形態素解析ができるようにする。

通常のdataflowのサンプルと異なるのはmecabで使う辞書ファイルをどのように配布するかという問題だ。今回は、パッケージ等は作らなず、インスタンスが生成された後、GCSにおいたmecabの辞書ファイルを読んでいる。

今回はtwitterで使われる人名は誰が多いのかをサンプルとして調べることにする。

shibacow@xxxx~$ mecab -d /usr/lib/x86_64-linux-gnu/mecab/dic/mecab-ipadic-neologd
初音ミクは歌う
初音ミク        名詞,固有名詞,人名,一般,*,*,初音ミク,ハツネミク,ハツネミク
は      助詞,係助詞,*,*,*,*,は,ハ,ワ
歌う    動詞,自立,*,*,五段・ワ行促音便,基本形,歌う,ウタウ,ウタウ
EOS

の

初音ミク        名詞,固有名詞, **人名** ,一般,*,*,初音ミク,ハツネミク,ハツネミク

に様に、mecabで人名と判断されるものを抜き出して人名を数える。neologdの人名辞書、例えば安倍晋三を人名として判断しないなど、ちょっと挙動がわからないところがある。

手順

neologdの辞書を用意
生成したneologdの辞書をGCSに設置
dataflowの実行

dataflowを利用して形態素解析を行う

今回作ったサンプルプログラムはこちらのgithubに公開している。

mecab-neologd の準備

こちらのサイトを参考に、mecab-neologdの用意をする。
見たところ、 pip でいれたmecab-python3をそのまま使い、辞書の場所を、

mecab = MeCab.Tagger('-r/dev/null -d 辞書ファイルの場所')

で指定すれば良いようだ。

GCSにneologdの辞書を置く

GCSのバケットを生成し、そこ先程作った辞書を置く。

私は次のファイルをおいた。

char.bin
dicrc
left-id.def
matrix.bin
pos-id.def
rewrite.def
right-id.def
sys.dic
unk.dic

dataflowの実行

次のrun.shを作り実行する

#!/bin/bash
export PROJECT=xxxxxxx <- あなたのプロジェクトID
export BUCKET=xxxxxxxxxx <- 辞書ファルをおいたり実行結果を入れたりするバケット

export REGION=us-central1
export RUNNER=DataflowRunner
export DT=`date "+%s"`
#export MODE=COST_OPTIMIZED
export MODE=SPEED_OPTIMIZED
export INPUT=$PROJECT:user_bq_dataset.user_bq_table
export MACHINE_TYPE=n1-highmem-16
#export MACHINE_TYPE=n1-standard-2
python3 mecab_count.py \
        --region $REGION \ <- リージョン BQのデータがUSの場合は、USのリージョンを指定する。
        --input  $INPUT \ <- BQのテーブル project_id:dataset.table_name のフォーマット 
        --output gs://$BUCKET/results_$DT/outputs \ <- GSのバケット
        --runner $RUNNER \ <- DataflowRunner
        --project $PROJECT \ <- Project name
        --flexrs_goal $MODE \ <- COST優先かスピード優先 
        --dicpath gs://$BUCKET/dic/mecab-ipadic-neologd \ <- GCSに設置した辞書の場所
        --machine_type  $MACHINE_TYPE \ <- インスタンスのタイプ
        --temp_location gs://$BUCKET/tmp/ \ <- tmpファイルの場所 GCSの場所を指定する
        --setup_file ./setup.py <- setup.pyにロードするライブラリを指定(mecabとか)

コードの説明

エントリーポイントになる mecab_count.py は単純に from mecab_word_count import mecab_word_count を呼び出すためのもの（後述するように、main文で、superが使えなかったためモジュールを作った）。

コードはこちら

もともとのapache beam のサンプル word_count の word_count.py を拡張した。

ポイント

setup

DoFnでレコードのエレメントごとに呼び出されるのは、process 関数になる。 setup は、インスタンスの生成時、もしくは少数しか呼ばれないためこの時点で、gcsから辞書のロードを行う。また、mecabのインスタンスの生成も行う。

processでやるとエレメントごとに呼ばれる為、大量に呼ばれてしまうが、setupであれば少数しか呼ばれない。

google cloud storage ライブラリではなく、beam付属のGcsIOを使う。

mecabの辞書ファイルをGCSからダウンロードする時に、ふつうであれば、 google-cloud-storage ライブラリを使うがどうもbeamのライブラリと相性が悪いようで実行時にエラーになる。
そのため、

gcs = gcsio.GcsIO()

のように、 apache_beam.io.gcp.gcsio をつかって、GCSから辞書ファイルをローカルにロードする。

        gcs = gcsio.GcsIO()
        src=gcs.open(gcs_path)
        with open(dstpath,'wb') as outf:
          outf.write(src.read())

のようにすれば、GCSからデータをロードして、ローカルに保存してくれる。

mecab前のインターバル

    time.sleep(120)
    path = os.getcwd()+os.sep+'dic/mecab-ipadic-neologd'
    self.mecab = MeCab.Tagger('-r/dev/null -d {}'.format(path))

setupの中で、time.sleepを呼んでいるのは、GCSでのロードの直後に、

    self.mecab = MeCab.Tagger('-r/dev/null -d {}'.format(path))

を呼ぶとインスタンスが大量に立ち上がったときに、エラーが起こったためのワークアラウンド（正しい処置ではない気がする）。

__parse

ストップワードとして、

ex_word=(u'RT',u'さん',u'様',u'ちゃん',u'くん',u'笑',u'氏',u'君',u'ツイ',u'さま')

を使った（人名としてmecabから出力されるが人名ではないため）、

実行結果

tweetされた年と、人名とペアにして次のような出力がGCSに保存される。

2018_初音ミク: 34508

コード部分はこちらになる。

# code
   for e in names:
      elm=u"{}_{}".format(created_at.year,e)
      result.append(elm)

年ごとの多い順のランキングを作ると次のようになる。
ローソンと、前澤以外は人名ではないような気がする。ジュジュンは人名かな？SixTONESはグループ名かな。

人気人名を2018,2019,2020,2021年で500名までを列挙した。
人気人名.csv

rank	2018	2019	2020	2021
1	…。	譲	譲	ローソン
2	譲	…。	求	譲
3	求	ローソン	…。	求
4	ローソン	求	ローソン	原
5	LINE	金	金	…。
6	な！	な！	な！	トク
7	金	LINE	リプ	金
8	ジェジュン	万名	LINE	前澤
9	チキ	ジェジュン	SixTONES	な！
10	かな	リプ	かな	LINE

例えば、初音ミクだと次のような順位になる

shibacow@dataflow-twitter:~/prog/dataflow/mecab_result$ ./collect_name.py --limit 50000 --name 初音ミク
INFO:root:y=2018 rank=270 data=year=2018 body=初音ミク cnt=34508
INFO:root:y=2019 rank=201 data=year=2019 body=初音ミク cnt=45339
INFO:root:y=2020 rank=138 data=year=2020 body=初音ミク cnt=63762
INFO:root:y=2021 rank=122 data=year=2021 body=初音ミク cnt=3704

2021年は1月20日まで。徐々にランクが上がっている。

dataflowを使う利点

PCollectionを使うことで、分散処理を透過的に実行可能
インスタンスを増やせば、処理数も増える。
一台のマシンで不可能でも、複数台並べてスケールできる

javaを使った同等の処理との比較

完全に同一ではないが、こちらの記事のようにjava-dataflow使って、tweetを分析をmecab-neologd-kuromoji で行っている。

java-dataflowでのCPU利用率

javaのdataflowはCPUの利用率が安定している。

python-dataflowでのCPU利用率

python-dataflowではCPU利用率にゆらぎがある

効率的にCPUを使うという点ではjavaのほうが分がありそうだが、pythonでは処理のロジックがシンプルなので、コードの書きやすさ、効率性のどちらを選ぶかを決めれば良いと思う。

はまった点

requestments.txtを指定するとそこからワーカーが動かない

setup.pyに移行すると上手く動いた。

main文でsuperが上手く出来ないため、別モジュールにきりだした

例えば、word count examle の WordExtractingDoFn に対して、


class WordExtractingDoFn(beam.DoFn):
   def __init__(self,foobar):
       super(WordExtractingDoFn, self).__init__()

などの拡張しても上手く実行されない。

AttributeError: Can't get attribute 'WordExtractingDoFn

などとエラーが出る。
理由は、こちらにあるが、

Beam does not support super() calls in main module on Python 3.

dataflowのpython3でmainでのsuperの使用が出来ないからのようだ。

bigqueryのデータを使う場合は、dataflowのインスタンスを立ち上げるリージョンに気をつけよう

bigqueryのデータは、us にあるそうだ。そのため、そのデータを、asiaでdataflowのインスタンスを立ち上げて利用すると、データの移動が発生するそのデータの移動でコストが結構かかる。dataflowをus-central1 で利用した方がコストが安いようだ。

下に添付した画像のように、転送コストの方がCPU利用コストより高かった。

Tips

FlexRS を活用するとコストカットできる

dataflowでは、最大6時間ほど持って実行する代わりに、コストが安くなるFlexRSのCOST_OPTIMIZED が使える。

FlexRSに関してはこちらを参照

私の実験では、7億件のtweetの処理通常であれば、21$ かかっていたが、FlexRSの COST_OPTIMIZEDモードであれば、 4.72$ で済んだ。通常であれば、全てレギュラーインスタンスを使うが、COST_OPTIMAIZED モードであれば、プリエンティティブVMを活用するため。

apache-beam-pythonを用いてGCP dataflowでmecabを使い形態素解析をする話

概要

手順