-
Notifications
You must be signed in to change notification settings - Fork 4.2k
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.
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
[BEAM-6553] A Python SDK sink that supports File Loads into BQ #7655
Conversation
Run Python Dataflow ValidatesRunner |
1 similar comment
Run Python Dataflow ValidatesRunner |
Run Python PostCommit |
cc: @reuvenlax |
Run Python PostCommit |
Run Python PreCommit |
Run Python PostCommit |
1 similar comment
Run Python PostCommit |
Fixed lint issue. Ready for review. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks Pablo. This looks great.
Sorry approved by mistake. Just wanted to send comments. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
(resending as a change request)
@chamikaramj I've addressed comments. LMK if that is reasonable. Overview of changes:
|
job_references = [elm[1] for elm in dest_ids_list] | ||
|
||
while True: | ||
status = self._check_job_states(job_references) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The BQ wrapper has retries. This is similarly handled for e.g. inserts[1][2]
[1] https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/gcp/bigquery.py#L635-L646
[2] https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/gcp/bigquery_tools.py#L578-L617
equal_to([job_reference]), label='CheckJobs') | ||
|
||
|
||
@unittest.skipIf(HttpError is None, 'GCP dependencies are not installed') |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@chamikaramj there are three integration tests added here. They test different kinds of functionality. They can be reduced to two, but I'd say two is the minimum.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Removed one IT. Two more are left with a number of unit tests.
Run Python PostCommit |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks!
_MAXIMUM_LOAD_SIZE = 15 * ONE_TERABYTE | ||
|
||
# Big query only supports up to 10 thousand URIs for a single load job. | ||
_MAXIMUM_SOURCE_URIS = 10*1000 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Where is this used ?
|
||
|
||
def _generate_file_prefix(pipeline_gcs_location): | ||
# If a gcs location is provided to the pipeline, then we shall use that. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: GCS
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hm? GCS is what it says : )
|
||
copy_job_name = '%s_copy_%s_to_%s' % ( | ||
job_name_prefix, | ||
_bq_uuid('%s:%s.%s' % (copy_from_reference.projectId, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will we end up creating the same UUID for the same destination table ? Will this result in jobs being rejected when we try to execute multiple jobs against the same destination table ? It'll be interesting to see what Java SDK currently does in this case.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There is now an index for jobs to the same table, and this id is not regenerated.
_bq_uuid('%s:%s.%s' % (copy_from_reference.projectId, | ||
copy_from_reference.datasetId, | ||
copy_from_reference.tableId)), | ||
_bq_uuid('%s:%s.%s' % (copy_to_reference.projectId, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should check if the job already exists in BQ and use that if the job exists (it's important to make sure that the BQ ID is unique for this Beam/Dataflow job instance)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The id is generated once, and not recomputed, so it's always the same for a single pipeline execution. We've resolved this in person.
table_reference.tableId = job_name | ||
yield pvalue.TaggedOutput(TriggerLoadJobs.TEMP_TABLES, table_reference) | ||
|
||
job_reference = self.bq_wrapper.perform_load_job( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should also check if the job exists before creating to be resilient to workitem/bundle failures.
|
||
@retry.with_exponential_backoff( | ||
num_retries=MAX_RETRIES, | ||
retry_filter=retry.retry_on_server_errors_and_timeout_filter) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should be careful about timeout filters. Timeout does not give any guarantees regarding whether the job was received by the BQ service or not.
Run Python PostCommit |
Please resolve the conflicts. |
Improving todos, documentation
Run Python PostCommit |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks Pablo. This looks great. LGTM.
Please squash commits and self merge.
apache#7655)" This reverts commit cdea885.
apache#7655)" This reverts commit cdea885.
This PR implements file loads into BQ from the Python SDK for Batch pipelines.
r: @chamikaramj
Open question
Features being implemented
WriteToBigQuery
Test results