GCPのBigQueryを使ってみました

BigQueryを使ってみたので、関連情報をまとめました

GCP NoSQL を使ってデータ処理したい時

主なサービスはこの辺りだと思います。

• BigQuery: 最も古い。DWH。格安。
• Google BigQuery でヒストリカルデータ保存の料金を半分に、 クエリの速度を 10 倍に
• BigTable: 新しい。リアルタイム分散系NoSQL。高い。
• Googleがスケーラビリティとパフォーマンスの高いNoSQLデータベースCloud Bigtableをベータで提供開始
• Datastore: 新しい。NoSQLフルマネージドかつ、少し安い。
• ウェブ、モバイル アプリケーションのためのスケーラブル NoSQL データベース
• Compute Engine/Container Engine:好きなものを構築できる。使い方によっては、安く済む。

今回は、格安で比較的枯れているBigQueryを使ってみました。 以降の記事はBigQueryについてです。

オライリー本もあります。 Google BigQuery

課金について

ビッグデータをCloudサービスで使う時、課金方式をよく把握しておくのは必須です。

• データ保存による: 1TB 1ヶ月 $20
• スキャンデータ量による
• BigQueryで150万円溶かした人の顔
• 300GBに対して、フルスキャンを数百回繰り返した。(テーブル切り分けのために、select insert)
• 価格体系が今後変わる。computing量による感じに。
• ストリーミングは課金: 10万行単位などバルクインサートより高い。
• 費用的には、Cloud Storageなどにデータを置いて、バルクインサートが良い。
• ストリーミングの方が手間は少ない(Fluentdや、以下のPandasなどで直接インサートできる)
• ストリーミングはデータ落ちなどが発生する模様。

BigQuery の課金仕様と注意点をまとめてみた（2015-07 時点）

課金対策ツール

• Table partitions Creating and Updating Date-Partitioned Tables
• Table decorator Table Decoratorを使って処理するレコード数を削減する
• テーブルサイズを小さく保つ

• Dry runを使って料金計算

  bq query --dry_run 'select * from table' | grep -o '[1-9][0-9]*' | awk '{print $1 / 1024 / 1024 / 1024 / 1024 * 5}'

• 同じクエリはキャッシュされて料金制限される
• カスタム割り当て機能を使う

• 料金予測=>即時反映されない

使い方

Pythonから使う場合がメインです。

データ分析基盤としての BigQuery 運用のベストプラクティス Python使いのためのBigQuery連携

また、SDKが提供されてない言語でも、REST APIから使えます。 BigQuery クライアント ライブラリ

その場合、OAUTH2認証が必要です。

CLIで使う

Googleが提供している、基本のbqコマンドです。

bq コマンドライン ツール クイックスタート

インタラクティブに使う

GCPのコンソールからのやり方です。

Quickstart Using the Web UI

Jupyter(IPython)ベースのインタラクティブツールもあります。
Betaですが、Jupyterはかなり使いやすいので、こちらもオススメです。

Google Cloud Datalab 上のページのCLOUD DATALAB を起動は使えなくなってる模様。立ち上げはこちらから。dockerを使って起動する。 quickstart

アプリで使う(ライブラリ)

• Pandas: io.gbqパッケージがあり、read_gbq、to_gbqが用意されてます。pandas使っている方ならこれ一択かと。かなり使いやすいです。
• Pythonのclientライブラリ: BigQuery-Pythonがあります。Pandas使ってない方はこちらが候補になると思います。使いやすいです。
• 公式API:上記でできないことがある場合以外、直接触ることはないです。ライブラリ作成者向けだと思います。

Pandas

to_gbqでinsertすると、ストリーミングになり、課金対象になります。
大量データを入れる場合は計算した方が良いです。

pandas連携が詳しく載ってるので再掲します。

Python使いのためのBigQuery連携 pandas.io.gbq.read_gbq pandas.io.gbq.to_gbq

Python client

BigQuery-Python PythonからBigQueryを操作するときは BigQuery-Python が便利だった

公式API

BigQuery API クイックスタート BigQuery API Client Library for Python getting_started.py Google Application Default Credentials

Google Compute Engine インスタンス間でやり取り(service account と mysql in Debian)

Google Compute Engine間でやり取りをするときの設定です。

gcloudを既存のインスタンスで使う

Service Accountsを作って、その認証情報をgcloudに設定します。

認証エラーが出ます

認証設定してない時は、このようなエラーが出ます。

$ gcloud compute instances list
ERROR: (gcloud.compute.instances.list) Some requests did not succeed:
 - Insufficient Permission

auth loginでセットアップしようとすると...

$ gcloud auth login

You are running on a Google Compute Engine virtual machine.
It is recommended that you use service accounts for authentication.

service accountsを使った方が良いと勧められます。

サービスアカウントを作成

今回は、コンソールからサービスアカウントを追加しました。 詳細手順は、公式ドキュメントのCONSOLEタブにあります。

Creating and Managing Service Accounts

gcloudの設定

上記で取得した、キーファイルをインスタンス上に設置して、アクティベイトします。

$ vi ~/.cred.json
$ chmod 400 ~/.cred.json
$ gcloud auth activate-service-account --key-file ~/.cred.json
$ gcloud compute instances list
...

参考： gcloud auth activate-service-account Creating and Enabling Service Accounts for Instances

MySQLをインスタンス間で接続する

MySQLを設定して、インスタンス間のネットワークを設定します。

とりあえずつなぐと疎通出来ませんでした。

$ telnet 10.240.0.4 3306
Trying 10.240.0.4...
telnet: Unable to connect to remote host: Connection refused

Compute Engine

• インスタンスのネットワーク設定を確認する
• デフォルトで、default-allow-internalがついてる。変えて無ければOK。

How to Set Up Remote Access to MySQL on Google Compute Engine Google Compute Engineのファイヤーウォールを理解する

mysqlの設定

クライアント側でmysql-clinetをインストール

sudo apt-get install mysql-client

Debianでは、セキュリティ上の理由から、MySQLサーバは、デフォルトでリモートアクセスを停止しています。

• /etc/my.cnfのbind-address=${instance-ip}に設定します

• mysqlをリスタートします。

  $ sudo systemctl restart mysql
  $ systemctl status mysql

mysql server

あとは、ユーザの権限を確認します

select user, host from mysq.user

確認

クライアント側のtelnetで疎通確認

$ telnet XXX 3306
Trying XXX...
Connected to XXX.
Escape character is '^]'.

クライアント側のmysqlコマンドで接続

mysql -u xxx -p -h xxx

pandasでの簡単な日付処理

pandasは、日付処理も充実しています。

すぐに使える関数をまとめました。

準備

import pandas as pd

日付に変換

Stringから作成する場合。

> pd.to_datetime('2016-08-26 00:00:00')
Timestamp('2016-08-26 00:00:00')

to_datetimeは、かなり賢く以下のような入力がデフォルトで正しくパースされます。

> pd.to_datetime('2016/8/26 0:0')
> pd.to_datetime('2016/8/26')
> pd.to_datetime('20160826')
> pd.to_datetime('160826 0000')
Timestamp('2016-08-26 00:00:00')

ただし、日付6桁は欧米式にパースされます。

> pd.to_datetime('260816 0100')
Timestamp('2016-08-26 01:00:00')

Python pandas で日時関連のデータ操作をカンタンに

日ごと、月ごと、年ごとに処理

年月日を取り出します。

> pd.to_datetime('2016-08-26 0:0:0').year
2016

> pd.to_datetime('2016-08-26 0:0:0').month
8

> pd.to_datetime('2016-08-26 0:0:0').day
26

dtを使うとdatetime64型の要素のプロパティに直接アクセスできます。

> df['date1']
0   2016-08-26
1   2016-08-27
Name: date1, dtype: datetime64[ns]

> df['date1'].dt.month
8

groupbyする場合です。

# 各カラムを月別に合計
df.groupby(df['date1'].dt.month).sum()

# グルーピングしたデータを取得
df.groupby(df['date1'].dt.month).groups

Python pandas アクセサ / Grouperで少し高度なグルーピング/集計

pandasでmysqlに接続する in debian

ライブラリインストールから、接続までまとめました。

mysql-connector-python + pandas.io.sqlです。

setup

リポジトリからが簡単に入ります。

$  git clone https://github.com/mysql/mysql-connector-python.git
$  cd mysql-connector-python
$  python ./setup.py build
$  sudo python ./setup.py install

condaを使ってる場合です。

$ conda install -c https://conda.anaconda.org/anaconda mysql-connector-python 

ディストリビューションからも取れる模様。

$ sudo apt-get install python3-mysql.connector

参考

python+pandasで大規模データを扱うときのメモ
Python3でMySQLに接続する環境をLinuxで整える
macでのmysql-connector-pythonインストール方法

DataFrameをテーブルから直接作る

import mysql.connector as mc
import pandas.io.sql as pdsql

con = mc.connect(
  user ='USER',
  password='PW',
  host='localhost',
  database='DB')

sql = 'SELECT * FROM table'

df = pdsql.read_sql(sql, con)
# ここでそのまま使えます。

con.close

参考

pandasでRDBの読み書きをする

Haskellでさくっとデータ処理をする

Haskellの強みの一つは、関数合成やパターンマッチを使って複雑なロジックをバグを抑えながら書けることだと思います。

さくっとデータ処理をやりたい時に使えそうなところをまとめました。

まずはシンプルに

ghciとdoctest

対話型のghciを使うと試しながらできるのでオススメです。

そこである程度固まってきたら、hsファイルにまとめたくなると思います。

doctestは、簡単な実行環境としても使えます。 Haskellの単体テスト最前線

Listを使う

列ごとにListとして取り込みます。

行は、Listをzipしてtupleを作ると柔軟に使えます。

Record構文は、さくっとやりたい時には、手間がかかります。

>>> id = ["a","b","c"]
>>> num = [1,2,3]
>>> data = zip id num
>>> data
[("a", 1), ("b", 2), ("c", 3)]

モナドについて

IOは出力に必要ですが、それ以外は最低限必要なものだけを使います。

実践的にモナドを使う方針と具体的な使い方が参考になります。

Haskellで競技プログラミング IO編

Maybeは、欠損値がある場合に使います。

型宣言しない

型宣言をつけると格好がつきますが、さくっとやる場合は、基本つけなくてもいいと思います。

少し実践的に

Vector

パフォーマンスを考えたらぜひ使いたいライブラリです。

List操作は楽ですが、パフォーマンスがネックになる場合があります。

その時に使えます。

また、データ処理系のライブラリでは、デフォルトで使われています。

https://github.com/haskell/vector

csvを使う

cassavaがシンプルに使えて便利です。

https://github.com/hvr/cassava

サクッと使うならば、ヘッダなしがオススメ。 レコードごとに処理するサンプルコードです。

{-# LANGUAGE ScopedTypeVariables #-}

import qualified Data.ByteString.Lazy as BL
import Data.Csv
import qualified Data.Vector as V

main :: IO ()
main = do
    csvData <- BL.readFile "salaries.csv"
    case decode NoHeader csvData of
        Left err -> putStrLn err
        Right v -> V.forM_ v $ \ (name, salary :: Int) ->
            putStrLn $ name ++ " earns " ++ show salary ++ " dollars"

ここで、decode NoHeader csvDataは、Vectorを返すので、forM_以外のVectorの関数にも慣れておくといいです。

joinを作ってみた

Haskellはロジック記述力が高いので、シンプルにいろいろと書けます。

SQLのjoinに相当する操作(結合条件を指定して、二つの表をつなげる)を探しましたが、意外とありませんでした。

実際にデータを処理するときは、キー結合したいときは結構多いため、書いてみました。

キーを条件に要素をつなげてますが、結合条件を関数で渡してもよかったと思います。 inner joinは、zip, filter, intersectなどでそのまま書けます。

import Data.List
import Data.Maybe as M

-- | Join 
--
-- Examples:
--
-- >>> fullJoin [1,2] [10,20] [2,3] [200, 300]
-- [(1,Just 10,Nothing),(2,Just 20,Just 200),(3,Nothing,Just 300)]
--
-- >>> leftJoin [1,2] [10,20] [2,3] [200, 300]
-- [(1,Just 10,Nothing),(2,Just 20,Just 200)]
--
-- >>> rightJoin [1,2] [10,20] [2,3] [200, 300]
-- [(2,Just 20,Just 200),(3,Nothing,Just 300)]
-- 
-- >>> innerJoin [1,2] [10,20] [2,3] [200, 300]
-- [(2,Just 20,Just 200)]
--
-- >>> data Row = Row { col1 :: Integer , col2 :: Integer } deriving (Eq, Show)
-- >>> type Table = [Row]
-- >>> let table = [Row {col1 = 10, col2 = 11}, Row{col1 = 20, col2 = 21}]
-- >>> fullJoin [1,2] table [2,3] [200,300]
-- [(1,Just (Row {col1 = 10, col2 = 11}),Nothing),(2,Just (Row {col1 = 20, col2 = 21}),Just 200),(3,Nothing,Just 300)]
--
-- >>> let { f i (Just x) Nothing = (i, col1 x, col2 x, 0); f i (Just x) (Just y) = (i, col1 x, col2 x, y); f i Nothing (Just y) = (i, 0, 0, y); f i Nothing Nothing = (i, 0, 0, 0); }
-- >>> fullJoinWith f [1,2] table [2,3] [200,300]
-- [(1,10,11,0),(2,20,21,200),(3,0,0,300)]

fullJoinWith f ks1 vs1 ks2 vs2 = map (\(i, x, y) -> f i x y) $ fullJoin ks1 vs1 ks2 vs2

leftJoin :: Ord k => [k] -> [a] -> [k] -> [b] -> [(k, M.Maybe a, M.Maybe b)]
leftJoin ks1 vs1 ks2 vs2 = _join ks1 ks1 vs1 ks2 vs2

rightJoin :: Ord k => [k] -> [a] -> [k] -> [b] -> [(k, M.Maybe a, M.Maybe b)]
rightJoin ks1 vs1 ks2 vs2 = map (\(x, y, z) -> (x, z, y)) $ _join ks2 ks2 vs2 ks1 vs1

fullJoin :: Ord k => [k] -> [a] -> [k] -> [b] -> [(k, M.Maybe a, M.Maybe b)]
fullJoin ks1 vs1 ks2 vs2 = _join (union ks1 ks2) ks1 vs1 ks2 vs2

innerJoin :: Ord k => [k] -> [a] -> [k] -> [b] -> [(k, M.Maybe a, M.Maybe b)]
innerJoin ks1 vs1 ks2 vs2 = _join (intersect ks1 ks2) ks1 vs1 ks2 vs2

_join :: Ord k => [k] -> [k] -> [a] -> [k] -> [b] -> [(k, M.Maybe a, M.Maybe b)]
_join ids ks1 vs1 ks2 vs2 = zipWith(\(x, y) (_, y2) -> (x, y, y2)) zs1 zs2
  where
    zs1 = mkZs ks1 vs1 
    zs2 = mkZs ks2 vs2
    mkZs ks vs = sortByFst . filterByFst . (++) (zipWith(\k v -> (k, Just v)) ks vs) $ zip (ids \\ ks) (repeat Nothing) 
    sortByFst = sortBy (\(x1, _) (x2, _) -> compare x1 x2)
    filterByFst = filter (flip elem ids . fst) 

tensorflow の sample を動かす

以下は、tensorflowの公式チュートリアルの一つ、手書き数字認識のサンプルです。

TensorFlow MNIST tutorial

インストール時にエラーが出る場合があります。 最新版を入れて、protobufを入れ直して解決しました。

インストール

このsampleを試す場合、tensorflowのインストールは、このやり方がオススメです。

少し古いバージョンを直接指定したところ、input_dataのパッケージパスの指定が、tutorialのやり方ではうまくいきませんでした。 また、最新が欲しい方は、githubからcloneする方法もあります。 https://github.com/tensorflow/tensorflow/issues/890

export TF_BINARY_URL=https://storage.googleapis.com/tensorflow/linux/cpu/tensorflow-0.10.0rc0-cp27-none-linux_x86_64.whl
pip install --upgrade $TF_BINARY_URL --trusted-host storage.googleapis.com

データインポート

tutorial に従ってデータをインポートすると以下のようなエラーが出ました。

> from tensorflow.examples.tutorials.mnist import input_data
...
ImportError: Traceback (most recent call last):
  File "/home/YutakaNishimura/tensorflow/lib/python2.7/site-packages/tensorflow/python/__init__.py", line 52, in <module>
      from tensorflow.core.framework.graph_pb2 import *
        File "/home/YutakaNishimura/tensorflow/lib/python2.7/site-packages/tensorflow/core/framework/graph_pb2.py", line 6, in <module>
            from google.protobuf import descriptor as _descriptor
            ImportError: No module named google.protobuf


Error importing tensorflow.  Unless you are using bazel,
you should not try to import tensorflow from its source directory;
please exit the tensorflow source tree, and relaunch your python interpreter
from there.

protobufを確認すると入ってました。
インストールし直しました。

pip uninstall protobuf
pip install protobuf

再度実行すると成功しました。

In [1]: from tensorflow.examples.tutorials.mnist import input_data

In [2]: mnist = input_data.read_data_sets('MNIST_data', one_hot=True)

Successfully downloaded train-images-idx3-ubyte.gz 9912422 bytes.
Extracting MNIST_data/train-images-idx3-ubyte.gz
Successfully downloaded train-labels-idx1-ubyte.gz 28881 bytes.
Extracting MNIST_data/train-labels-idx1-ubyte.gz
Successfully downloaded t10k-images-idx3-ubyte.gz 1648877 bytes.
Extracting MNIST_data/t10k-images-idx3-ubyte.gz
Successfully downloaded t10k-labels-idx1-ubyte.gz 4542 bytes.
Extracting MNIST_data/t10k-labels-idx1-ubyte.gz

Pandasの基本的でよく使いそうな機能のメモ

Pandasの基本的ですが、実際によく使いそうな使い方のメモです。

インストールについて

こちらが非常に詳しいです。 anacondaを使って環境構築します。

Google Compute Engine + Debian で実行しましたが、問題なく動いてます。

データサイエンティストを目指す人のpython環境構築 2016

csv形式のStringからデータを読み取る。

Stringから直接csvデータを読み込むこともできます。 http://stackoverflow.com/questions/22604566/how-to-create-a-pandas-dataframe-from-string

import sys
from io import StringIO

d = StringIO("""a,b,c
1,2,3
11,12,13
""")

df = pd.read_csv(d)

データフレームを連結する

Python Pandasでのデータ操作の初歩まとめ − 前半：データ作成＆操作編 pandasでindexの連番を振り直す

concatもしくは、append関数を使います。

行を追加する場合も、DataFrameの連結で行います。

• concatは対象とするDataFrameをリストにして渡します。
• appendはdf.appendという書き方になるので、直接行を追加する場合に便利です。
• どちらもデフォルトでは、非破壊的な動作になっています。

行方向に追加

concatを使って連結

// axis=0 (デフォルト。省略可能)
concated_df = pd.concat([df1, df2])

ちなみに、カラム名が異なると、カラムは全て残り、互いの欠損箇所には、NA値が挿入されます。

欠損値を埋めたい場合は、fillnaを使います。統計処理で必要とされる、平均値補完、線形補間なども簡単にできます。

// 0埋め
df.fillna(0)
// 平均値
df.fillna(df.mean())
// 線形補間
df.interpolate()
// na値削除(行ベース)
df.dropna()
// na値削除(列ベース)
df.dropna(axis=1)

appendを使って行をその場で追加

// columnsを既存のものから取ってきて同じ構造にします
appended_df = df.appned(pd.DataFrame([[1,2,3]], columns = df.columns.tolist())

列方向に追加

pd.concat([df1, df2], axis=1)

同じくインデックスが違うと列方向にNA値が挿入される。

新しい列をその場で追加する方法は、こちらが簡単です。 ただ、dfを直接書き換えるので注意が必要です。

df['c_new'] = [1,2,3]

インデックス・カラムの貼り直し

行番号の貼り直し

concated_df.reset_index(drop=True)

//直接生成時に貼り直すこともできる
pd.concat([df1,df2]).reset_index(drop=True)

カラム名の貼り直し

// dfインスタンスに対して設定
concated_df.columns = ["c1","c2","c3","c4","c5","c6"]

表記ブレをなくすよう変更

// 全て小文字にして、スネークケースにする。記号も開きました。
df.columns = [s.lower().replace(' ', '_').replace('?', '_bool') for s in df.columns.tolist()]

SQLのような処理を行う

よくまとまっていてオススメです。

pandasにsqlでよくやる処理をやらせてみる

sort

sort_valuesを使う方が推奨のようです。

df.sort_values(by="col1")

ERROR

CSVのパースエラー

CSVを読み込む時に、パースエラーが発生しました。

> df = pd.read_csv("file.csv")

pandas/parser.pyx in pandas.parser.TextReader.read (pandas/parser.c:8748)()

pandas/parser.pyx in pandas.parser.TextReader._read_low_memory (pandas/parser.c:9003)()

pandas/parser.pyx in pandas.parser.TextReader._read_rows (pandas/parser.c:9731)()

pandas/parser.pyx in pandas.parser.TextReader._tokenize_rows (pandas/parser.c:9602)()

pandas/parser.pyx in pandas.parser.raise_parser_error (pandas/parser.c:23325)()

CParserError: Error tokenizing data. C error: EOF inside string starting at line 244276

ここで line 244276 は、ファイル上の行数でないので注意が必要です。

この時は結果的に、csvファイルの最終行のデータがおかしくなっていためエラーになっていました。

それでも解決しない場合は、この辺りが参考になると思います。

ISSUEが立っていますが、この問題は現在は、フィックスされてるよう。 https://github.com/pydata/pandas/issues/5501

カラムが多すぎるとExceptionを吐くが、それらを削除して読み込む。

pd.read_csv("file.csv", error_bad_lines=False)

エンコーディングとエンジンを指定する

pd.read_csv("file.csv", encoding='utf-8', engine='c')
pd.read_csv("file.csv", encoding='utf-8', engine='python')

quotingを指定する

import csv
pd.read_csv("file.csv", quoting=csv.QUOTE_NONE)

文字列として読み込んで解析する

f = open("file.csv")
s = f.read()
l = f.readlines()