Skip to content

[AIRFLOW-1389] BigQueryOperator should support createDisposition #2470

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yu-iskw wants to merge 2 commits into from
Closed

[AIRFLOW-1389] BigQueryOperator should support createDisposition #2470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8985dc0
[AIRFLOW-1389] BigQueryOperator should support `createDisposition`
yu-iskw Jul 22, 2017
b35f887
Change the order of parameters
yu-iskw Jul 27, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion airflow/contrib/hooks/bigquery_hook.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,7 +195,8 @@ def run_query(
allow_large_results=False,
udf_config = False,
use_legacy_sql=True,
maximum_billing_tier=None):
maximum_billing_tier=None,
create_disposition='CREATE_IF_NEEDED'):
"""
Executes a BigQuery SQL query. Optionally persists results in a BigQuery
table. See here:
Expand All @@ -210,6 +211,9 @@ def run_query(
BigQuery table to save the query results.
:param write_disposition: What to do if the table already exists in
BigQuery.
:type write_disposition: string
:param create_disposition: Specifies whether the job is allowed to create new tables.
:type create_disposition: string
:param allow_large_results: Whether to allow large results.
:type allow_large_results: boolean
:param udf_config: The User Defined Function configuration for the query.
Expand Down Expand Up @@ -238,6 +242,7 @@ def run_query(
configuration['query'].update({
'allowLargeResults': allow_large_results,
'writeDisposition': write_disposition,
'createDisposition': create_disposition,
'destinationTable': {
'projectId': destination_project,
'datasetId': destination_dataset,
Expand Down
13 changes: 11 additions & 2 deletions airflow/contrib/operators/bigquery_operator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,12 @@ class BigQueryOperator(BaseOperator):
(<project>.|<project>:)<dataset>.<table> that, if set, will store the results
of the query.
:type destination_dataset_table: string
:param write_disposition: Specifies the action that occurs if the destination table
already exists. (default: 'WRITE_EMPTY')
:type write_disposition: string
:param create_disposition: Specifies whether the job is allowed to create new tables.
(default: 'CREATE_IF_NEEDED')
:type create_disposition: string
:param bigquery_conn_id: reference to a specific BigQuery hook.
:type bigquery_conn_id: string
:param delegate_to: The account to impersonate, if any.
Expand Down Expand Up @@ -61,12 +67,14 @@ def __init__(self,
udf_config=False,
use_legacy_sql=True,
maximum_billing_tier=None,
create_disposition='CREATE_IF_NEEDED',
*args,
**kwargs):
super(BigQueryOperator, self).__init__(*args, **kwargs)
self.bql = bql
self.destination_dataset_table = destination_dataset_table
self.write_disposition = write_disposition
self.create_disposition = create_disposition
self.allow_large_results = allow_large_results
self.bigquery_conn_id = bigquery_conn_id
self.delegate_to = delegate_to
Expand All @@ -81,5 +89,6 @@ def execute(self, context):
conn = hook.get_conn()
cursor = conn.cursor()
cursor.run_query(self.bql, self.destination_dataset_table, self.write_disposition,
self.allow_large_results, self.udf_config, self.use_legacy_sql,
self.maximum_billing_tier)
self.allow_large_results, self.udf_config,
self.use_legacy_sql, self.maximum_billing_tier,
self.create_disposition)