Datalabで簡易webクローリングし、BigQueryにアップロードする方法

#はじめに
GCP Cloud Datalab(Python3)で、requestsライブラリを利用してcsvを特定サイトから取得し、取得したデータをBigQueryに格納するコードを説明します。

Datalabで簡易webクローリングし、GoogleCloudStorage(GCS)にアップロードする方法、のちょっと修正版です。

こんなユーザが対象です。

• ローカルでJupyterNotebookを使っていたが、クラウドでいつでもどこからでも利用したい
• WEBのデータ(csv)を簡単に取得してBQへアップロードして、機械学習や可視化したい

##環境

• Cloud Datalab(Python3)
• 読み込むsample.csvは以下のようなものでテストします。

date	memo	number
2018-10-08	test1	100
2018-10-09	test2	200
2018-10-10	test3	300

Datalab上でのPythonコード詳細

まず、必要なライブラリ読み込みます。storage,bigquery。。

import requests
import sys
import google.datalab.storage as storage
import google.datalab.bigquery as bq

適宜、変数を定義します。

## バケットの定義、適宜変更

## バケットの定義
bucket_name = '<bucket name>' // バケット名
bucket_path = 'gs://' + bucket_name //バケットのPATH
bucket_dir = '<bucket directory>/' //GCSの保存先フォルダ名
upload_filename = 'sample_upload.csv' //GCSに保存する名前
bucket = storage.Bucket(bucket_name)

## アクセスするURL
access_url = 'https://<replace to site>/sample.csv' //ダウンロードしたいサイトのcsv

## BigQuery
datasetname = '<bigquery datasets name>' // 保存先のBigQueryのデータセット名
table_id = '<bigquery table name' // BigQueryのテーブル名

対象のURLからrequestsを利用して、ファイルを取得。response.contentにコンテンツが格納されます。typeはbytes。

## URLからダウンロードする  
def download(url):
  data = ""
  response = requests.get(url)
  #print(type(response.content))
  ## <class 'bytes'>
  
  if response.status_code == 200:
    data = response.content
    return data
  else:
    print('download error')
    exit()
    return data

GCSにアップロードする関数は以下です。返り値に、objectのURIを返します(gs://形式)。

## GCSにアップロードする
def upload_gcs(bucket, uploadObject, data):
  upload_object = bucket.object(uploadObject)
  upload_object.write_stream(data,content_type='text/csv')
  return upload_object.uri

BQにデータを格納する関数は以下です。スキーマを定義しておき、load関数でGCSにあるデータをBQにinsertします。

## BigQueryに格納する
def upload2bq(file_path,id):
    schema = [
        {'name': 'date', 'type': 'DATE'},
        {'name': 'memo', 'type': 'STRING'},
        {'name': 'number', 'type': 'INT64'}
    ]
    ## csvオプション設定、最初の行をスキップして、urf-8形式とします。
    csv_option = bq.CSVOptions(skip_leading_rows=1, encoding=u'utf-8')    
    bq_schema = bq.Schema.from_data(schema)
    tablename = datasetname + '.' + id
    bq_table = bq.Table(tablename).create(schema = bq_schema, overwrite = True)

    ## GCSのpathを指定して、BQにデータをinsertします。
    bq_table.load(file_path, mode='append', source_format = 'csv', csv_options=csv_option)

最後に、main関数は以下です。実行後、BigQueryに、新たなテーブルが作成されるとともに、データが格納されます。

# main
if __name__ == "__main__":
  ## アクセスする先のURL
  url = access_url

  ## バケットインスタンスの作成
  bucket = storage.Bucket(bucket_name)
  
  ## URLにGETアクセスした結果のコンテンツを取得
  data = download(url)
  if data != "":
    ## 正常に取得できたら、指定したGCSへアップロードし、返り値として、URIを取得する(filepath)
    filepath = upload_gcs(bucket, bucket_dir + upload_filename, data)
    table_id = table_id
    upload2bq(filepath,table_id)
    print('Success!')
  else:
    print('not Success!')

参考サイト

*https://googledatalab.github.io/pydatalab/google.datalab.bigquery.html

おわりに

Datalabを利用して、簡単に、WEBから取得したcsvファイルを、BQに格納する方法を説明しました。変数は自分の環境用に変更をして試してみてください。