diff --git a/guide/17-working-with-knowledge-graphs/part4_create_replicas_and_synchronize_changes.ipynb b/guide/17-working-with-knowledge-graphs/part4_create_replicas_and_synchronize_changes.ipynb new file mode 100644 index 0000000000..a78c8353fb --- /dev/null +++ b/guide/17-working-with-knowledge-graphs/part4_create_replicas_and_synchronize_changes.ipynb @@ -0,0 +1,522 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "15b7f95a", + "metadata": {}, + "source": [ + "## Create Replicas and Synchronize Changes" + ] + }, + { + "cell_type": "markdown", + "id": "673b5f3f", + "metadata": {}, + "source": [ + "

Table of Contents

\n", + "
\n", + " \n", + "
" + ] + }, + { + "cell_type": "markdown", + "id": "7d9256d5", + "metadata": {}, + "source": [ + "## Introduction\n", + "\n", + "The [Sync feature](https://developers.arcgis.com/rest/services-reference/enterprise/kgs-sync-overview/) on Knowledge Graph Services allows you to define and create replicas and sync changes to and from the Knowledge Graph Service, depending how the replica has been defined.\n", + "\n", + "In order to use the Sync feature, it must be enabled on your Knowledge Graph Service. To enable it, either:\n", + "1. For ArcGIS Enterprise 12.2 and later, navigate to the portal item for the Knowledge Graph Service. In the Settings section, find the 'Enable Sync' option, turn it on, and save changes.\n", + "2. For ArcGIS Enterprise 12.1, navigate to the admin rest endpoint for the Knowledge Graph Service (example: https://myportal.com/server/rest/admin/services/Hosted/myKnowledgeGraph/KnowledgeGraphServer), click the 'Update Feature' option, input `{\"supportsSync\": true}` and click updateFeature to apply the change.\n", + "\n", + "Note: The Sync feature is only available on ArcGIS managed Knowledge Graph Services on ArcGIS Enterprise 12.1 or later.\n", + "\n", + "Once Sync is enabled, you can start to define, create, and sync replicas." + ] + }, + { + "cell_type": "markdown", + "id": "51d23677", + "metadata": {}, + "source": [ + "## Create Replica" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "69d5cda5", + "metadata": {}, + "outputs": [], + "source": [ + "# import all the classes that will be needed\n", + "from arcgis.graph import (\n", + " KnowledgeGraph, \n", + " EntityTypeReplicaDefinition, \n", + " RelationshipTypeReplicaDefinition, \n", + " ReplicaFilters, \n", + " StaticDefinition, \n", + " ReplicaRequest, \n", + " Entity, \n", + " SynchronizeReplicaRequest, \n", + " SyncDownloadParameters, \n", + " ReplicaDeltas, \n", + " SyncUploadEdits, \n", + " SyncApplyEdits \n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5939d0f5", + "metadata": {}, + "outputs": [], + "source": [ + "# set up entity type replica definitions\n", + "plant_def = EntityTypeReplicaDefinition(\n", + " type_name = \"Plant\" # entire type\n", + ")\n", + "utility_def = EntityTypeReplicaDefinition(\n", + " type_name = \"Utility\", \n", + " open_cypher_where_clause=\"n.Utility_Name='Bloom Energy'\" # filtered type (note: property filters are a beta feature)\n", + ")\n", + "\n", + "# set up relationship type replica definitions\n", + "owns_def = RelationshipTypeReplicaDefinition(type_name = \"owns\") # entire type\n", + "\n", + "# set replica filters using the definitions\n", + "replica_filters = ReplicaFilters(\n", + " static_definition = StaticDefinition(\n", + " entity_type_replica_definitions = [plant_def,utility_def], \n", + " relationship_type_replica_definitions=[owns_def]\n", + " )\n", + ")\n", + "\n", + "# create replica using a replica request\n", + "create_result = knowledge_graph.create_replica(\n", + " ReplicaRequest(\n", + " replica_definition = replica_filters, \n", + " replica_name = \"replica_withfilters_bidirectional\", \n", + " direction = \"BIDIRECTIONAL\"\n", + " )\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "d9f88186", + "metadata": {}, + "source": [ + "## Use Replica Data\n", + "\n", + "The response from the replica creation contains a couple different important elements:\n", + "- errors: any errors from the request\n", + "- warnings: any warnings from the request\n", + "- replica_id: id assigned to the replica\n", + "- replica_name: name of the replica, this could be different from the requested name which can be found in requested_name\n", + "- sync_date: datetime the replica was created (we will use this later)\n", + "- replica_data: the actual data from the service based on the replica definitions, we will use this to create a replica-based service" + ] + }, + { + "cell_type": "markdown", + "id": "1f0783d3", + "metadata": {}, + "source": [ + "[BETA] There is a beta function that allows you to easily create a new knowledge graph service using the replica data from creating a replica. This should be used with caution and is subject to change in a future release." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1bf3189e", + "metadata": {}, + "outputs": [], + "source": [ + "replica_based_kg, new_replica_id, new_replica_sync_date = create_result.replica_data._create_graph_from_replica(gis=gis, name=\"service_from_replica_data\")" + ] + }, + { + "cell_type": "markdown", + "id": "5d258650", + "metadata": {}, + "source": [ + "Since this function is currently beta and may not work for all cases yet, the process for completing what is done in the function includes:\n", + "1. Create a new knowledge graph service, this is where the replica data will be added" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9aef5d80", + "metadata": {}, + "outputs": [], + "source": [ + "# set any additional features that should be supported\n", + "create_props = {\"supportsProvenance\": True, \"supportsEditorTracking\": False, \"supportsSync\": True}\n", + "\n", + "# create the new service and connect to it\n", + "new_kg_item = gis.content.create_service(name=\"\", service_type=\"KnowledgeGraph\", create_params={\"name\": \"new_service_name\",\"capabilities\": \"Query,Editing,Create,Update,Delete\",\"jsonProperties\": create_props})\n", + "replica_based_kg = KnowledgeGraph(new_kg_item.url, gis=gis)" + ] + }, + { + "cell_type": "markdown", + "id": "d08283e8", + "metadata": {}, + "source": [ + "2. Use the `service_data_model` in the `create_result.replica_data` to populate all data model information (types, properties, indexes, domains, etc) that will be needed in the new service.\n", + "\n", + "System maintained properties will be populated by the server and need to be dropped from the types in order to add them without errors." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "7cd2298c", + "metadata": {}, + "outputs": [], + "source": [ + "service_dm = create_result.replica_data.service_data_model\n", + "all_system_maintained_properties = {} # collect system maintained properties, these will be used during data editing as well\n", + "\n", + "for ent_type in service_dm.entity_types:\n", + " system_maintained_properties = []\n", + " # remove any types that are not regular\n", + " if (ent_type.role != \"esriGraphNamedObjectRegular\"):\n", + " service_dm.entity_types.remove(ent_type)\n", + " continue\n", + " props_to_remove = []\n", + " # remove system maintained properties\n", + " for prop in ent_type.properties:\n", + " if prop.is_system_maintained == True:\n", + " system_maintained_properties.append(prop.name)\n", + " props_to_remove.append(ent_type.properties.index(prop))\n", + " for prop_idx in sorted(props_to_remove, reverse=True):\n", + " del ent_type.properties[prop_idx]\n", + " indexes_to_remove = []\n", + " # remove system maintained indexes\n", + " for index in ent_type.field_indexes:\n", + " if 'esri__' in index.name.lower():\n", + " indexes_to_remove.append(ent_type.field_indexes.index(index))\n", + " for index_idx in sorted(indexes_to_remove, reverse=True):\n", + " del ent_type.field_indexes[index_idx]\n", + " # add all system maintained properties to the dictionary for editing\n", + " all_system_maintained_properties[ent_type.name] = system_maintained_properties\n", + "for rel_type in service_dm.relationship_types:\n", + " system_maintained_properties = []\n", + " # remove any types that are not regular\n", + " if (rel_type.role != \"esriGraphNamedObjectRegular\"):\n", + " service_dm.relationship_types.remove(rel_type)\n", + " continue\n", + " props_to_remove = []\n", + " # remove system maintained properties\n", + " for prop in rel_type.properties:\n", + " if prop.is_system_maintained == True:\n", + " system_maintained_properties.append(prop.name)\n", + " #system_maintained_properties.remove('globalid')\n", + " props_to_remove.append(rel_type.properties.index(prop))\n", + " for prop_idx in sorted(props_to_remove, reverse=True):\n", + " del rel_type.properties[prop_idx]\n", + " indexes_to_remove = []\n", + " # remove system maintained indexes\n", + " for index in rel_type.field_indexes:\n", + " if 'esri__' in index.name.lower():\n", + " indexes_to_remove.append(rel_type.field_indexes.index(index))\n", + " for index_idx in sorted(indexes_to_remove, reverse=True):\n", + " del rel_type.field_indexes[index_idx]\n", + " # add all system maintained properties to the dictionary for editing\n", + " all_system_maintained_properties[rel_type.name] = system_maintained_properties\n", + "# add all of the entity and relationship types now that they are prepared\n", + "types_add = replica_based_kg.named_object_type_adds(entity_types=service_dm.entity_types, relationship_types=service_dm.relationship_types, as_dict=False)\n", + "# print the result to check that it was successful\n", + "print(types_add)" + ] + }, + { + "cell_type": "markdown", + "id": "633e408f", + "metadata": {}, + "source": [ + "3. Use the `TYPE_ADDS` edits in the `replica_data` (this will be all of them if you are using the result of `create_replica()` but most likely not if you are using a delta-based download from `synchronize_replica()`)\n", + "\n", + "The list of system maintained properties from the last step will be used here to remove those properties (except globalid) from the edits to avoid errors from server." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a945e6e7", + "metadata": {}, + "outputs": [], + "source": [ + "entity_adds = []\n", + "relationship_adds = []\n", + "# get the replica data frames stream\n", + "generator = create_result.replica_data.stream_data_frames()\n", + "while True:\n", + " try:\n", + " replica_data_frame = next(generator)\n", + " # only use adds\n", + " if (replica_data_frame.edit_type == \"TYPE_ADD\"):\n", + " # iterate through each data frame\n", + " items = replica_data_frame.iterate_data()\n", + " for item in items:\n", + " # get properties to delete based on system-maintained status and delete them from each instance\n", + " delete_me = []\n", + " for prop in item[0].properties.keys():\n", + " if prop in all_system_maintained_properties[item[0].type_name]:\n", + " delete_me.append(prop)\n", + " for i in delete_me:\n", + " del item[0].properties[i]\n", + " # create separate entity and relationship adds lists\n", + " if type(item[0]) == Entity:\n", + " entity_adds.append(item[0])\n", + " if type(item[0]) == Relationship:\n", + " relationship_adds.append(item[0])\n", + " except StopIteration:\n", + " break\n", + "# add entities first so they are there for the relationships\n", + "entity_response = replica_based_kg.apply_edits(adds=entity_adds, as_dict=False)\n", + "print(entity_response)\n", + "# add relationships which will attach to the created entities\n", + "relationship_response = replica_based_kg.apply_edits(adds=relationship_adds, as_dict=False)\n", + "print(relationship_response)" + ] + }, + { + "cell_type": "markdown", + "id": "57102e4b", + "metadata": {}, + "source": [ + "4. Set up and create a replica that contains all types so we can get changes made when we are ready to synchronize." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3bbf3c39", + "metadata": {}, + "outputs": [], + "source": [ + "replicakg_replica_filters = ReplicaFilters(\n", + " static_definition=StaticDefinition(\n", + " entity_type_replica_definitions=[\n", + " EntityTypeReplicaDefinition(type_name=entity_type.name)\n", + " for entity_type in service_dm.entity_types\n", + " ],\n", + " relationship_type_replica_definitions=[\n", + " RelationshipTypeReplicaDefinition(type_name=relationship_type.name)\n", + " for relationship_type in service_dm.relationship_types\n", + " ],\n", + " )\n", + ")\n", + "\n", + "# create replica using a replica request\n", + "replica_create_result = replica_based_kg.create_replica(\n", + " ReplicaRequest(\n", + " replica_definition = replicakg_replica_filters,\n", + " replica_name = \"replica_fulltypes_bidirectional\",\n", + " direction = \"BIDIRECTIONAL\"\n", + " )\n", + ")\n", + "\n", + "new_replica_id = replica_create_result.replica_id\n", + "new_replica_sync_date = replica_create_result.sync_date" + ] + }, + { + "cell_type": "markdown", + "id": "80ba4ca7", + "metadata": {}, + "source": [ + "Now we have the `replica_based_kg`, `new_replica_id`, and `new_replica_sync_date` from either of the two paths (beta function or step-by-step)." + ] + }, + { + "cell_type": "markdown", + "id": "3f8d7e30", + "metadata": {}, + "source": [ + "## Make Edits To New Service\n", + "\n", + "Any edits made on services with Sync enabled will be tracked, then we can use `synchronize_replica()` to get the changes and sync them back to the parent Knowledge Graph Service." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "7ce215ae", + "metadata": {}, + "outputs": [], + "source": [ + "# make some edits in the replica kg with sync on so changes are tracked\n", + "replica_based_kg.apply_edits(\n", + " adds=[\n", + " Entity(\n", + " type_name=\"Plant\", \n", + " properties={\"Plant_Name\":\"new_plant\"}\n", + " )\n", + " ]\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "60f8c411", + "metadata": {}, + "source": [ + "## Download Changes\n", + "\n", + "To get all edits made from the time we created the replica on the new service, `synchronize_replica()` can be used to download the replica deltas based on the `sync_date` property available on the replica request reponse." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d467ea39", + "metadata": {}, + "outputs": [], + "source": [ + "replica_sync_download_response = replica_based_kg.synchronize_replica(\n", + " SynchronizeReplicaRequest(\n", + " replica_id=new_replica_id, \n", + " download_parameters=SyncDownloadParameters(\n", + " replica_deltas=ReplicaDeltas(\n", + " last_sync_date=new_replica_sync_date.timestamp()*1000\n", + " )\n", + " )\n", + " )\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "aabf7e04", + "metadata": {}, + "source": [ + "## Upload Changes\n", + "\n", + "The replica data in the download response can then be uploaded to the parent Knowledge Graph Service that the original replica was created on.\n", + "\n", + "[BETA] There is a beta function `_update_from_graph_replica()` that simplifies the process of updating the original knowlege graph service from the replica-based service changes." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "edfd9c01", + "metadata": {}, + "outputs": [], + "source": [ + "knowledge_graph._update_from_graph_replica(create_result.replica_id,replica_sync_download_response.sync_data)" + ] + }, + { + "cell_type": "markdown", + "id": "1a2972bd", + "metadata": {}, + "source": [ + "The steps completed in that beta function can also be done manually, they are:\n", + "1. Create the list of edits for each edit type from the downloaded sync data" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "dd3c65eb", + "metadata": {}, + "outputs": [], + "source": [ + "# create adds, updates, deletes from replica data\n", + "def generate_edit_types_lists(sync_data):\n", + " adds = []\n", + " updates = []\n", + " deletes = []\n", + " df_generator = sync_data.stream_data_frames()\n", + " for replica_df in df_generator:\n", + " print(replica_df.edit_type)\n", + " if replica_df.edit_type == \"TYPE_ADD\":\n", + " for edit in replica_df.iterate_data():\n", + " adds.append(edit[0])\n", + " if replica_df.edit_type == \"TYPE_UPDATE\":\n", + " for edit in replica_df.iterate_data():\n", + " updates.append(edit[0])\n", + " if replica_df.edit_type == \"TYPE_DELETE\":\n", + " for edit in replica_df.iterate_data():\n", + " deletes.append(edit[0])\n", + " return {\"adds\": adds, \"updates\": updates, \"deletes\": deletes}" + ] + }, + { + "cell_type": "markdown", + "id": "2214e2d6", + "metadata": {}, + "source": [ + "2. Use those edits for `synchronize_replica()` uploads to the knowledge graph service." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "0bdc8db5", + "metadata": {}, + "outputs": [], + "source": [ + "all_edits = knowledge_graph.generate_edit_types_lists(replica_sync_download_response.sync_data)\n", + "knowledge_graph.synchronize_replica(\n", + " SynchronizeReplicaRequest(\n", + " replica_id=create_result.replica_id,\n", + " upload_edits=SyncUploadEdits(\n", + " sync_apply_edits=SyncApplyEdits(\n", + " adds=all_edits[\"adds\"],\n", + " updates=all_edits[\"updates\"],\n", + " deletes=all_edits[\"deletes\"]\n", + " )\n", + " )\n", + " )\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "a55584d0", + "metadata": {}, + "source": [ + "## Check Service Updates\n", + "\n", + "Finally, now that the changes that were downloaded have been uploaded to the original service a simple query can be done to check if the changes made it there properly." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c049402d", + "metadata": {}, + "outputs": [], + "source": [ + "result = knowledge_graph.query_streaming(\"MATCH (p:Plant) WHERE p.Plant_Name CONTAINS 'new_plant' RETURN p\", as_dict=False)\n", + "list(result)" + ] + } + ], + "metadata": { + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +}