You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/11/07 11:59:22 UTC
[GitHub] [beam] vchunikhin commented on a diff in pull request #23800: [Playground][Backend] Dataset processing with python scripts to deploy examples
vchunikhin commented on code in PR #23800:
URL: https://github.com/apache/beam/pull/23800#discussion_r1015325709
##########
playground/infrastructure/datastore_client.py:
##########
@@ -83,54 +87,67 @@ def save_to_cloud_datastore(self, examples_from_rep: List[Example], sdk: Sdk, or
examples_ids_before_updating = self._get_all_examples(sdk, origin)
# loop through every example to save them to the Cloud Datastore
- with self._datastore_client.transaction():
- for example in tqdm(examples_from_rep):
- sdk_key = self._get_key(DatastoreProps.SDK_KIND, Sdk.Name(example.sdk))
- example_id = self._make_example_id(origin, sdk, example.name)
- updated_example_ids.append(example_id)
- examples.append(
- self._to_example_entity(example, example_id, sdk_key, actual_schema_version_key, origin)
- )
- snippets.append(
- self._to_snippet_entity(example, example_id, sdk_key, now, actual_schema_version_key, origin)
- )
- pc_objects.extend(self._pc_object_entities(example, example_id))
- files.append(self._to_file_entity(example, example_id))
-
- self._datastore_client.put_multi(examples)
- self._datastore_client.put_multi(snippets)
- self._datastore_client.put_multi(pc_objects)
- self._datastore_client.put_multi(files)
-
- # delete examples from the Cloud Datastore that are not in the repository
- examples_ids_for_removing = list(filter(lambda key: key not in updated_example_ids, examples_ids_before_updating))
- if len(examples_ids_for_removing) != 0:
- logging.info("Start of deleting extra playground examples ...")
- examples_keys_for_removing = list(
- map(lambda ex_id: self._get_key(DatastoreProps.EXAMPLE_KIND, ex_id), examples_ids_for_removing)
- )
- snippets_keys_for_removing = list(
- map(lambda ex_id: self._get_key(DatastoreProps.SNIPPET_KIND, ex_id), examples_ids_for_removing)
- )
- file_keys_for_removing = list(
- map(self._get_files_key, examples_ids_for_removing)
+
+ for example in tqdm(examples_from_rep):
+ sdk_key = self._get_key(DatastoreProps.SDK_KIND, Sdk.Name(example.sdk))
+ example_id = self._make_example_id(origin, sdk, example.name)
+ updated_example_ids.append(example_id)
+ examples.append(
+ self._to_example_entity(example, example_id, sdk_key, actual_schema_version_key, origin)
+ )
+ snippet = self._to_snippet_entity(example, example_id, sdk_key, now, actual_schema_version_key, origin)
+ pc_objects.extend(self._pc_object_entities(example, example_id))
+ files.append(self._to_file_entity(example, example_id))
+ if example.datasets and example.emulators:
+ dataset = example.datasets[0]
+ emulator = example.emulators[0]
+ file_name = f"{dataset.name}.{dataset.format}"
+ link = self._upload_dataset_to_bucket(file_name)
+ dataset = self._to_dataset_entity(file_name, link)
+ datasets.append(dataset)
+ dataset_nested_entity = self._to_dataset_nested_entity(file_name, example_id, emulator)
+ snippet_datasets = [dataset_nested_entity]
+ snippet.update(
+ {
+ "datasets": snippet_datasets
+ }
)
- pc_objs_keys_for_removing = []
- for example_id_item in examples_ids_for_removing:
- for example_type in [
- PrecompiledExample.GRAPH_EXTENSION.upper(),
- PrecompiledExample.OUTPUT_EXTENSION.upper(),
- PrecompiledExample.LOG_EXTENSION.upper()
- ]:
- pc_objs_keys_for_removing.append(
- self._get_key(DatastoreProps.PRECOMPILED_OBJECT_KIND,
- f"{example_id_item}{config.DatastoreProps.KEY_NAME_DELIMITER}{example_type}")
- )
- self._datastore_client.delete_multi(examples_keys_for_removing)
- self._datastore_client.delete_multi(snippets_keys_for_removing)
- self._datastore_client.delete_multi(file_keys_for_removing)
- self._datastore_client.delete_multi(pc_objs_keys_for_removing)
- logging.info("Finish of deleting extra playground examples ...")
+ snippets.append(snippet)
+
+ if datasets:
+ self._datastore_client.put_multi(datasets)
+ self._datastore_client.put_multi(examples)
+ self._datastore_client.put_multi(snippets)
+ self._datastore_client.put_multi(pc_objects)
+ self._datastore_client.put_multi(files)
+
+ # delete examples from the Cloud Datastore that are not in the repository
+ examples_ids_for_removing = list(filter(lambda key: key not in updated_example_ids, examples_ids_before_updating))
+ if len(examples_ids_for_removing) != 0:
+ logging.info("Start of deleting extra playground examples ...")
+ examples_keys_for_removing = list(
+ map(lambda ex_id: self._get_key(DatastoreProps.EXAMPLE_KIND, ex_id), examples_ids_for_removing)
+ )
+ snippets_keys_for_removing = list(
+ map(lambda ex_id: self._get_key(DatastoreProps.SNIPPET_KIND, ex_id), examples_ids_for_removing)
+ )
+ file_keys_for_removing = list(
+ map(self._get_files_key, examples_ids_for_removing)
+ )
+ pc_objs_keys_for_removing = []
+ for example_id_item in examples_ids_for_removing:
+ for example_type in [
+ PrecompiledExample.GRAPH_EXTENSION.upper(),
+ PrecompiledExample.OUTPUT_EXTENSION.upper(),
+ PrecompiledExample.LOG_EXTENSION.upper()
+ ]:
+ pc_objs_keys_for_removing.append(
+ self._get_key(DatastoreProps.PRECOMPILED_OBJECT_KIND, f"{example_id_item}{config.DatastoreProps.KEY_NAME_DELIMITER}{example_type}"))
+ self._datastore_client.delete_multi(examples_keys_for_removing)
+ self._datastore_client.delete_multi(snippets_keys_for_removing)
+ self._datastore_client.delete_multi(file_keys_for_removing)
+ self._datastore_client.delete_multi(pc_objs_keys_for_removing)
+ logging.info("Finish of deleting extra playground examples ...")
Review Comment:
I'd create a separate task for that. It looks that we are not able to delete unused datasets here because python scripts are executed for every SDK separately but a dataset can relate to an example from other SDK. Thus, we can compare links that store in GCS and links in _pg_datasets_ kind by a scheduled task in golang to remove them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org