Skip to content

Commit eaca17d

Browse files
committed
Fix bq load for clashing copy job names
1 parent 8d252c4 commit eaca17d

2 files changed

Lines changed: 10 additions & 5 deletions

File tree

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
{
22
"comment": "Modify this file in a trivial way to cause this test suite to run.",
33
"pr": "37345",
4-
"modification": 49
5-
}
4+
"modification": 50
5+
}

sdks/python/apache_beam/io/gcp/bigquery_file_loads.py

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,13 @@ def _bq_uuid(seed=None):
170170
else:
171171
return str(hashlib.md5(seed.encode('utf8')).hexdigest())
172172

173+
def _bq_uuid_list(list):
174+
checksum = hashlib.sha256()
175+
for item in list:
176+
checksum.update(item.encode('utf-8'))
177+
# separator
178+
checksum.update(b'\x00')
179+
return checksum.hexdigest()
173180

174181
class _ShardDestinations(beam.DoFn):
175182
"""Adds a shard number to the key of the KV element.
@@ -589,9 +596,7 @@ def process(
589596
write_disposition = 'WRITE_APPEND'
590597
wait_for_job = False
591598

592-
chunk_job_name = copy_job_name_base
593-
if len(chunks) > 1:
594-
chunk_job_name = f"{copy_job_name_base}_{i}"
599+
chunk_job_name = '%s_%s' % (copy_job_name_base, _bq_uuid_list(chunk))
595600

596601
_LOGGER.info(
597602
"Triggering copy job %s from %s to %s (write_disposition: %s)",

0 commit comments

Comments
 (0)