0

bigquery-pythonを使って、ツイッターのデータをビッグクエリにインポートしようとしているのですができません。
どうやら、create_tableができていないみたいです。問題の部分だけ抽出すると以下のコードです。

import bigquery

# 認証情報はcredentials.jsonから取得。
# グーグルクラウドプラットフォームでサービスアカウントを作成して、「新しい秘密鍵の提供」から取得。
client = bigquery.get_client(json_key_file='credentials.json', readonly=False)

client.create_dataset(dataset_id='tiwtter')  # Falseが返ってくる。datasetは作成されていない。

環境:
OS :Ubuntu 14.04.5 LTS
python:3.6.4
conda :4.3.34
bigquery:1.14.0

念の為以下にコードの全文を記載します。

import os
import sys
from datetime import timezone

import tweepy
import bigquery
import pandas as pd

# Twitterの認証情報を読み込む。
twitter_api = pd.read_csv('api_twitter.csv', names=['name', 'value'], header=None)
twitter_api
CONSUMER_KEY = twitter_api.loc[0, 'value']
CONSUMER_SECRET = twitter_api.loc[1, 'value']
ACCESS_TOKEN = twitter_api.loc[2, 'value']
ACCESS_TOKEN_SECRET = twitter_api.loc[3, 'value']

auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

# BigQueryの認証情報(credentials.json)を指定してBigQueryのクライアントを作成する。
# 明示的にreadonly=Falseとしないと書き込みができない。
client = bigquery.get_client(json_key_file='credentials.json', readonly=False)

DATASET_NAME = 'twitter'  # BigQueryのデータ・セット名
TABLE_NAME = 'tweets'  # BigQueryのテーブル名

# テーブルが存在しない場合は作成する。
if not client.check_table(DATASET_NAME, TABLE_NAME):
        print('Creating table {0}.{1}'.format(DATASET_NAME, TABLE_NAME), file=sys.stderr)
        # create_table()の第三引数にはスキーマを指定する。
        client.create_table(DATASET_NAME, TABLE_NAME, [
            {'name': 'id',          'type': 'string',    'description': 'ツイートのID'},
            {'name': 'lang',        'type': 'string',    'description': 'ツイートの言語'},
            {'name': 'screen_name', 'type': 'string',    'description': 'ユーザー名'},
            {'name': 'text',        'type': 'string',    'description': 'ツイートの本文'},
            {'name': 'created_at',  'type': 'timestamp', 'description': 'ツイートの日時'},
        ])


class MyStreamListener(tweepy.streaming.StreamListener):
    """
    Streaming APIで取得したツイートを処理するクラス
    """

    status_list = []
    num_imported = 0

    def on_status(self, status):
        """
        ツイートを受信した時に呼び出されるメソッド。
        引数: ツイートを表すStatusオブジェクト
        """
        self.status_list.append(status)  # Statusオブジェクトをstatus_listに追加する。

        if len(self.status_list) >= 500:
            # status_listに500件溜まったらBigQueryにインポートする。
            if not push_to_bigquery(self.status_list):
                #  インポートに失敗した場合はFalseが帰ってくるのでエラーを出して終了する。
                print('Failed to send to bigquery', file=sys.stderr)
                return False

            # num_importedを増やして、status_listを空にする。
            self.num_imported += len(self.status_list)
            self.status_list = []
            print('Imported {0} rows'.format(self.num_imported), file=sys.stderr)

            # 料金が高額にならないように、5000件をインポートしたらFalseを返して終了する。
            # 継続的にインポートしたいときは次の2行をコメントアウトしてください。
            if self.num_imported >= 5000:
                return False


def push_to_bigquery(status_list):
    """
    ツイートのリストをBigQueryにインポートする。
    """

    # TweepyのStatusオブジェクトのリストからdictのリストに変換する。
    rows = []
    for status in status_list:
        rows.append({
            'id': status.id_str,
            'lang': status.lang,
            'screen_name': status.author.screen_name,
            'text': status.text,
            # datetimeオブジェクトをUTCのPOSIXタイムスタンプに変換する。
            'created_at': status.created_at.replace(tzinfo=timezone.utc).timestamp()
        })

    # dictのリストをBigQueryにインポートする。
    # 引数は順に、データ・セット名、テーブル名、行のリスト、行を一意に識別する列名を表す。
    # insert_id_keyはエラーでリトライした時に重複しないようにするために使われるが、必須ではない。
    return client.push_rows(DATASET_NAME, TABLE_NAME, rows, insert_id_key='id')


    # Stream APIの読み込みを開始する。
    print('Collecting tweets...', file=sys.stderr)
    stream = tweepy.Stream(auth, MyStreamListener())
    # 公開されているツイートをサンプリングしたストリームを受信する。
    # 言語を指定していないので、あるゆる言語のツイートを取得できる。
    stream.sample()
3
  • 使っているキーのサービスアカウントにBigQueryの操作に必要な役割を付与できていますか? Commented 2018年5月3日 6:26
  • ご指摘受けた部分できていませんでした。ただ、役割に「BigQuery管理者」を付与して新たにサービスアカウントを作成して、上記のコードを走らせてみたのですがやはり同じ結果で、テーブル作成できませんでした・・・。コメントいただけると幸いです。よろしくお願いします Commented 2018年5月3日 12:49
  • キーが問題なのか、プログラムが問題なのかわからないので、公式マニュアルのとおり1回してみたらどうですか。 cloud.google.com/bigquery/docs/reference/… Commented 2018年5月3日 13:30

1 件の回答 1

0

BigQueryの操作に必要な役割が付与されていない可能性が高いです。

公式マニュアルのとおり、クライアント ライブラリのインストールと環境変数の設定をして、

pip install --upgrade google-cloud-bigquery
export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

例を実行してみてください。

from google.cloud import bigquery

bigquery_client = bigquery.Client()
dataset_id = 'my_new_dataset'
dataset_ref = bigquery_client.dataset(dataset_id)
dataset = bigquery.Dataset(dataset_ref)

dataset = bigquery_client.create_dataset(dataset)
print('Dataset {} created.'.format(dataset.dataset_id))

公式のライブラリーだとエラーメッセージが出るので原因がわかります。

bigquery-pythonは、一見便利そうだけど、Falseしか返ってこないのは問題ですね。

1
  • 解決しました!!ありがとうございます。!!上記のコードを実行したら、 'Billing has not been enabled for this project.' と出てきました。請求先の設定ができていないプロジェクトだったことが原因でした・・・。トライアルに登録する前に、BigQueryのサンプルデータ触っている時に作成したプロジェクトをそのまま使っていました。長い時間お付き合いいただき本当にありがとうございました。 Commented 2018年5月4日 4:41

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。