CSV形式のオープンデータを Bluemix の Cloudant へロードするメモ

Cloudant に大量のデータをロードしようとして、いろいろな問題を解決したメモです。 Watson など AI 技術を駆使したシステムを開発する場合、正確なデータを沢山蓄えているほど、面白いとか、価値の高いAIとなっていきます。 そこで、オープンデータに注目して、非構造化データを取り扱えるデータベースへの取り込みを考えてみました。

実施したい事

CSV形式のオープンデータを自分のPCに取り込み、Excelで確認や編集などを実施して、CSV形式で保存して、Cloudant へロードする操作を簡単に実施したい。 アプリサーバーは Linux のため、SJISからUTF-8への漢字コード変換、改行コードの変更も一度に実行したい。 という主旨のものです。

オープンデータとは

オープンデータには、いろいろあるのですが、今回のサンプルコードでは、参考資料(1)の郵便局の郵便番号データをCloudantへロードしてみます。

オープンデータを取り扱っているウェブサイトから、オープンデータに関する説明文を引用します。

オープンデータ（Open Data）とは、特定のデータが、一切の著作権、特許などの制御メカニズムの制限なしで、全ての人が望むように利用・再掲載できるような形で入手できるべきであるというアイデアである。

参考資料(2)

日本政府は、公共データを広く公開することにより、国民生活の向上、企業活動の活性化等を通じ、我が国の社会経済の発展に寄与する観点から、機械判読に適したデータ形式を、営利目的も含めた二次利用が可能な利用ルールで公開する「オープンデータ」の取組を推進しています。

参考資料(3)

オープンデータとは，「機械判読に適したデータ形式で、二次利用が可能な利用ルールで公開されたデータ」であり「人手を多くかけずにデータの二次利用を可能とするもの」です。つまり，誰でも許可されたルールの範囲内で自由に複製・加工や頒布などができるデータをいいます。もちろん商用としても利用可能です。
「人口統計」や「公共施設の場所」などをはじめとした様々な公共のデータを、ユーザ(市民、民間企業など)に有効活用していただき、社会経済全体の発展に寄与することを目的として、世界中で同様の試みに取り組まれています。

参考資料(4)

オープンデータに関する日本政府の取り組みは、参考資料(5)にありますので、合わせて参照すると理解が深まると思います。

Cloudant とは

なぜ、Cloudant を選んで利用するかについて、以下に記載します。

IBM Cloudant は、アプリケーションとデータベースの間で高速で中断されることのないデータ・フローを実現する、管理不要の NoSQL JSONデータベース・サービスです。Cloudantを利用すれば、より付加価値の高いアプリを開発し、ビジネスを成長させ、さらに管理の手間も減らすことができるでしょう

参考資料(6)

この参考資料によると、次の様な特徴があります。

• Cloudant の水平スケーリング・アーキテクチャーは、数百万名のユーザーと数テラバイトのデータに対応することができ、ビジネスに合わせてシームレスに拡張できます。 ユーザーはそのユーザーに最も近いデータ・コピーに接続されるため、クラウド・ネットワークのオーバーヘッドに起因するデータ・アクセス遅延を抑えることができます。
• Cloudant の RESTful API により、データベース内のすべての文書に JSON としてアクセスできます。また、CloudantはApache CouchDB と互換性があるため、多数の言語ライブラリーやツールを利用できます。
• Cloudant には、iOS と Android の両方に対応したモバイル同期ライブラリー、Cloudant Syncが組み込まれています。Cloudant Sync によって構築されたオフライン・ファースト・アプリケーションは、オフライン、オンラインに関わらず優れたユーザー・エクスペリエンスを提供します。
• 地理空間オペレーションにおいて、境界指定の方法として単純な長方形以外の形状もサポートされており、 Web アプリケーションやモバイル・アプリケーションを差別化することができます。

さらに、Bluemix から利用する Cloudant Liteプランでは、1G未満のデータ量では、無料で使える点が大きいです。 参考資料(7)

Cloudant には、 Twitterもあり、@IBMCludant https://twitter.com/ibmcloudant で情報を収集することもできます。

事前準備事項

このメモを実際に実行するには、Bluemix のアカウントが必要です。 また、Bluemix にログイン後に、Cloudant をサービスを作成して、サービス資格情報を参照して、vcap-local.json.sample をコピーして vcap-local.json を作成して、credentials の情報をセットすることで、自分のノートPCなどの開発環境から、Bluemix上のCloudantをアクセスできる様になります。

Cloudant へのローディング・プログラム

下記の項目を編集して、利用できるPythonのプログラムを GitHUB https://github.com/takara9/cloudant_loader に置きました。 具体的な編集方法は、README.md を参照してください。

• Cloudant サービス資格情報
• 入力CSVファイル名
• Cloudant データベース名

ローディングプログラムの解説

この記事を読んでくれた人が、私のGitHubに登録したプログラムを参考に改造してもらえる様に、プログラムの解説を書いて置きます。

インプットとアウトプットの指定

実行する前に、output_db と input_file を指定します。 input_file は、DOS形式、SJIS そのままのファイル形式で良いです。 output_dbは、cloudant のデータベース名です。

cloudant_load_data.py

import sys
import json
import codecs
import uuid
import time
from cloudant.client import Cloudant
from cloudant.query import Query

#==================================
# ロード先データベース名
output_db = 'fukuoka'  <-- ココと

# インプットCSVファイル名
input_file = '40fukuoka.csv'  <-- ココ
output_file = input_file + ".utf-8"
#==================================

Cloudant のデータベースに、ロードされると、次の画面で確認することができます。 fukuoka として表示されている部分が、上記設定でロードされた結果になります。

この画面を表示するには、Bluemix メニュー -> サービス -> ダッシュボード -> Cloudant -> LAUNCH としてメニューを進んで行きます。

Cloudant への接続

JSON形式のサービス資格情報を読み込んで、Cloudant データベースへ接続します。 JSONファイルのフォーマットは、GitHub に、vcap-local.json.sample というファイル名で置いてありますので参考にして、作成してください。

cloudant_load_data.py

# Cloudant認証情報の取得
f = open('./vcap-local.json', 'r')
cred = json.load(f)
f.close()

print "クラウダントへの接続"
client = Cloudant(cred['services']['cloudantNoSQLDB'][0]['credentials']['username'], 
                  cred['services']['cloudantNoSQLDB'][0]['credentials']['password'], 
                  url=cred['services']['cloudantNoSQLDB'][0]['credentials']['url'])
client.connect()

サービス資格情報は、次の様にして参照できます。 Bluemix の Cloudant サービスの画面で、 Credentaial-1 の資格情報の表示をクリックすると、JSON形式のデータが表示されますから、コピペして利用します。

データベースの削除と作成

差分でロードするケースであれば、既存のデータベースを削除しないことで、追加として動作します。 一方、使えるデータとなるまで、何度か再読み込みが必要なケースでは、削除と作成を繰り返す事になりますので、次の様に書いてあります。

cloudant_load_data.py

# DBが存在していたら削除
print "既存データベース ", output_db ," の作成"
try:
    db = client[output_db]
    if db.exists():
        client.delete_database(output_db)
except:
    pass


# DBを新規作成
print "新規データベース ", output_db ," の作成"
try: 
    db = client.create_database(output_db)
    print "データベース作成成功"
except:
    print "データベース作成失敗"
sys.exit()

DOS形式からUNIX形式への変換

この部分で、SJISからUTF-8への漢字コード変換、そして、改行コードの変換して、中間ファイルを生成しています。
ここで、iconv を利用しない理由は、オープンデータの中に、SJISの範囲外のコードが含まれている場合、iconvではエラー無しに、処理を中断することがあるため、データロードに失敗していても気づかない恐れがあるためです。

このコードでは、try - except で変換エラーを補足して、CSVデータの行番号と表示していますので、変換に失敗した部分を書き残して、先に進めることができます。

cloudant_load_data.py

# 文字コード変換など前処理結果を中間ファイルへ出力
fin = codecs.open(input_file, "r", "shift_jis")
fout = codecs.open(output_file, "w", "utf-8")
line_no = 0
while True:
    try:
        line = fin.readline()
        if line == "":
            break
        line = line.rstrip('\r\n') + "\n"
        line = line.replace('"','')
        fout.write(line)
        line_no = line_no + 1
    except Exception as e:
        print "Line = ", line_no, " Error message: ", e
        pass

JSON形式の文書データを作り、Cloudantへロード

前述の漢字コード変換処理で、エラーで中断となる様な漢字コードを排除できたので、今度は速度重視で、"reader.readlines()" を利用して、全行を一度にメモリへ取り込みます。そして、一行一行 JSON文書を作成して、Cloudant へ登録して行きます。

# エクセルシートからのCSVデータを読んでDBへ登録する
reader = codecs.open(output_file, 'r', 'utf-8')
lines = reader.readlines()
print lines[0]

wait_cnt = 0
line_no = 0
for line in lines:

CSVファイルの１行目に、項目名行がある場合スキップする様にしています。 もし項目行が無い場合は、if line_no から始まる 3 行をコメントにするとよいでしょう。 それから、デバック表示を出しておくと進捗が把握できます。 次に、wait_cnt から始まる 4行ですが、これは 無料のCloudant Liteプランでは、書き込み速度に制限があり、秒間10件を超える書き込みが発生すると、エラーを返してきます。 その制限速度を守り、エラー発生を回避するために、9件書き込んだら 1秒間 WAIT を入れる様にしています。

    # ヘッダ行をスキップ
    if line_no == 0:
        line_no = line_no + 1
        continue

    # デバック表示
    print line_no,line.rstrip('\n')

    # 配列へ展開
    al = line.split(',')
    line_no = line_no + 1

    # Cloudant Liteプランの書き込み速度制限に対応
    wait_cnt = wait_cnt + 1
    if wait_cnt > 9:
        wait_cnt = 0
        time.sleep(1)

json形式の文書データとして、json_docを生成して、db.create_documentで書き込みを実施します。 ここで、JSONの項目名は、取り込もうとするCSVデータの項目に合わせて、命名する必要があります。 json_doc の書き方は、対象とするCSVによって、カスタマイズする必要があります。　このコードの最後に２行がコメントになっていますが、これは、書き込み確認の処理です。 この処理が入ると遅くなるので、コメントにしています。

    # DOC作成と書込み
    id = str(uuid.uuid4())
    json_doc = {
        #"_id": id,
        "_id": al[0],
        "jusho_CD": al[0],
        "todofuken_CD": al[1],
	　　　< 中略 >
        "jigyosyo_mei_kana": al[19],
        "jigyosyo_jyusyo": al[20]
        #"sin_jyusyo_kana": al[21]
     }

    # 書込み
    cdoc = db.create_document(json_doc)
    #if cdoc.exists():
    #    print "SUCCESS!!"

Cloudant の場合、_id の項目が無い場合、内部的にuuid を生成して、登録します。 しかし、一意のキーで取り出した場合は、_id に、キーをセットすることで、高速に取り出す事ができます。 ここでは、住所コードが一意のコードなので、_id にセットしています。

    json_doc = {
        #"_id": id,
        "_id": al[0],

まとめ

RDBを使ったシステム開発を実施している頃は、データのロードは、最初につまずく難関でした。 Cloudant を利用したからといって、そのデータロードが難関である事は、然程変わらない感じがしますね。

参考資料

(1) 郵便局 郵便番号データ ダウンロード http://www.post.japanpost.jp/zipcode/dl/kogaki-zip.html
(2) ウィキペディア オープンデータ https://ja.wikipedia.org/wiki/%E3%82%AA%E3%83%BC%E3%83%97%E3%83%B3%E3%83%87%E3%83%BC%E3%82%BF
(3) OPEN DATA JAPAN データカタログサイト http://www.data.go.jp/
(4) 自治体オープンデータ https://www.open-governmentdata.org/
(5) 総務省 オープンデータ戦略の推進 オープンデータとは http://www.soumu.go.jp/menu_seisaku/ictseisaku/ictriyou/opendata/opendata01.html
(6) IBM Cloudant (日本語サイト) https://www.ibm.com/analytics/jp/ja/technology/cloud-data-services/cloudant/
(7) IBM Bluemix カタログ Cloudant https://console.bluemix.net/catalog/services/cloudant-nosql-db?env_id=ibm:yp:us-south