google-bigquery - 如何在 Apache Airflow 中查询 Google Big Query 并将结果作为 Pandas Dataframe 返回？

Question

我正在尝试将 bigquery 查询保存到自定义 Airflow 运算符中的数据帧。

我试过使用airflow.contrib.hooks.bigquery_hook 和get_pandas_df 方法。任务卡在身份验证上，因为它希望我手动访问 url 进行身份验证。

结果，我在身份验证中进行了硬编码。这有效，但绝对不理想。

工作但不理想(凭证是硬编码的):

def execute(self, context):
        os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = 'my-file-location.json'
        client = bigquery.Client()

        job_config = bigquery.QueryJobConfig()

        df = client.query(
            self.query,
            location="US",
            job_config=job_config,).to_dataframe()

不工作:

def execute(self, context):
    bq  = BigQueryHook(bigquery_conn_id=self.gcp_conn_id, delegate_to=None,use_legacy_sql=True, location='US')
    df = bq.get_pandas_df(self.query)

此代码卡在身份验证中。这是日志:[2019-06-19 12:56:05,526] {logging_mixin.py:95} 信息 - 请访问此 URL 以授权此应用程序。

Answer 1

不知何故我无法得到BigQueryPandasConnector在职的。我最终得到的是使用 BigQueryHook 的凭据来创建一个普通的 bigquery.client.Client使用 BigQuery 的官方 Python 客户端。

下面是一个例子:

from google.cloud import bigquery

bq_hook = BigQueryHook(bigquery_conn_id=bigquery_conn_id, use_legacy_sql=False)
bq_client = bigquery.Client(project = bq_hook._get_field("project"), credentials = bq_hook._get_credentials())
df = bq_client.query(sql).to_dataframe()