ArcGIS API for Python | Esri Developer

Create Replicas and Synchronize Changes

Introduction

The Sync feature 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.

In the API, there are many new classes that have been added to make the replica creation and synchronization of changes possible. The general workflow will include:

Creating Replicas
- EntityTypeReplicaDefinition and/or RelationshipTypeReplicaDefinition: these allow you to define which types in the knowledge graph service you would like to be part of the replica that gets created. Within each type, you can also define specific property filters based on an openCypher where clause, note that this feature is currently in beta.
- ReplicaFilters: this is where you will define which sets of entity and relationship type replica definitions will be used.
- ReplicaRequest: takes your replica filters, a name for your replica, and the direction the replica will work in (uploads, downloads or bidirectional) and this will be passed into create_replica().
Synchronizing Changes
- SyncDownloadParameters and/or SyncUploadEdits: define what and how you would like to synchronize changes, downloading edits from a certain last sync date or uploading a set of edits.
- SynchronizeReplicaRequest: used to define the set of downloads and/or uploads to perform on the knowledge graph service.
Using Replica Data
- The ReplicaData class is used to access and work with replica data returned from either creating a replica or synchronizing changes.

The details of how to use these and the steps in between for handling replica data are outlined in this guide.

In order to use the Sync feature, it must be enabled on your Knowledge Graph Service. To enable it, either:

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.
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.
Use the code snippet below to enable sync on an existing knowledge graph service.

Note: The Sync feature is only available on ArcGIS managed Knowledge Graph Services on ArcGIS Enterprise 12.1 or later.

Once Sync is enabled, you can start to define, create, and sync replicas.

# get the portal item
knowledge_graph_item = gis.content.search("knowlege graph portal item name",item_type="Knowledge Graph")[0]

# enable sync on existing knowledge graph service
admin_url = knowledge_graph_item.url.replace(
    "/rest/services/",
    "/rest/admin/services/",
)
response = knowledge_graph_item._gis._con.post(
    f"{admin_url}/updateFeature",
    postdata={
        "supportsSync": True,
    },
)

Create Replica

There are many classes that will be used during the replication and synchronization process, all of those used in this guide are shown here. They can be split into sections to better understand how they are used.

General: KnowledgeGraph and Entity are general knowledge graph classes that will be used for initializing and making edits to the knowledge graphs.
Replica Creation: EntityTypeReplicaDefinition, RelationshipTypeReplicaDefinition, ReplicaFilters, StaticDefinition, and ReplicaRequest are all used together to define and create replicas on knowledge graph services.
Synchronize Changes: SynchronizeReplicaRequest, SyncDownloadParameters, ReplicaDeltas, SyncUploadEdits, and SyncApplyEdits are all used together to define what will be either downloaded or uploaded when synchronizes changes to a replica.

# import all the classes that will be needed
from arcgis.graph import (
    KnowledgeGraph,
    Entity,
    EntityTypeReplicaDefinition, 
    RelationshipTypeReplicaDefinition, 
    ReplicaFilters, 
    StaticDefinition, 
    ReplicaRequest, 
    SynchronizeReplicaRequest, 
    SyncDownloadParameters, 
    ReplicaDeltas, 
    SyncUploadEdits, 
    SyncApplyEdits 
)

EnttiyTypeReplicaDefinition and RelationshipTypeReplicaDefinition are a very important part of defining your replica request, these determine exactly what data from the knowledge graph service the replica will consist of. In this example, we are defining two EntityTypeReplicaDefinitions and one RelationshipTypeReplicaDefinition.

The first EntityTypeReplicaDefinition will include all Plant entities. The second will include only the Utility entities that match the filter defined in the open_cypher_where_clause.

The RelationshipTypeReplicaDefinition simply defines the type name of 'owns'. Since there is a filter defined on the Utility entities, only 'owns' relationships between Plant entities and Utilities that match the filter will be part of the replica data.

# set up entity type replica definitions
plant_def = EntityTypeReplicaDefinition(
    type_name = "Plant" # entire type
)
utility_def = EntityTypeReplicaDefinition(
    type_name = "Utility", 
    open_cypher_where_clause="n.Utility_Name='Bloom Energy'" # filtered type (note: property filters are a beta feature)
)

# set up relationship type replica definitions
owns_def = RelationshipTypeReplicaDefinition(type_name = "owns") # entire type

# set replica filters using the definitions
replica_filters = ReplicaFilters(
    static_definition = StaticDefinition(
        entity_type_replica_definitions = [plant_def,utility_def], 
        relationship_type_replica_definitions=[owns_def]
    )
)

# create replica using a replica request
create_result = knowledge_graph.create_replica(
    ReplicaRequest(
        replica_definition = replica_filters, 
        replica_name = "replica_withfilters_bidirectional", 
        direction = "BIDIRECTIONAL"
    )
)

Use Replica Data

The response from the replica creation contains a couple different important elements:

errors: any errors from the request
warnings: any warnings from the request
replica_id: id assigned to the replica
replica_name: name of the replica, this could be different from the requested name which can be found in requested_name
sync_date: datetime the replica was created (we will use this later)
replica_data: the actual data from the service based on the replica definitions, we will use this to create a replica-based service

Replica data can be used in many different ways, such as creating feature services, but for this example we'll see how it could be used to create a replica-based knowledge graph service.

The two main ways to retrieve replica data to work with are through create_replica() and synchronize_changes(). You can also access replica data through a zip file.

ReplicaData has a lot of information within it, including:

service_data_model: The data model for the service that the replica data is based on.
sync_date: Timestamp for the sync date of the replica data.
header: Contains spatial reference and transform information.
entity_types: List of the entity types included in the replica data along with their edit types.
relationship_types: List of the relationship types included in the replica data along with their edit types.
stream_data_frames(): This is what you will use to access the data frames and can be filtered using:
- types_names: Any specific entity or relationship types to include.
- type_categories: Can be used to filter only entities or only relationships.
- edit_types: The type of edit, this is particularly useful when using the synchronize_changes() response since there can be any mix of edit types. These include Add, Update, and Delete.

Each replica data frame retrieved from stream_data_frames() has additional information:

type_category: entity or relationship
type_name: name of the entity or relationship type
edit_type: add, update, or delete
iterate_data(): gathers all records within the data frame

# access replica data from create_replica()
replica_data = create_replica_response.replica_data

# access replica data from synchronize_replica()
replica_data = synchronize_replica_response.sync_data

# access replica data from a file
replica_data = ReplicaData(file_path)

# use the replica data
df_generator = replica_data.stream_data_frames() # generator to process all data frames
filtered_df_generator = replica_data.stream_data_frames(edit_types=['Update','Delete']) # filtered data frames based on edit types

for replica_df in df_generator:
    replica_df_data = replica_df.iterate_data()
    for data in replica_df_data:
        # do something with the data, these will be Entity and/or Relationship objects
        print(data)

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.

For the individual steps that are part of _create_graph_from_replica() see the end of this guide.

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")

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).

Make Edits To New Service

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.

# make some edits in the replica kg with sync on so changes are tracked
replica_based_kg.apply_edits(
    adds=[
        Entity(
            type_name="Plant", 
            properties={"Plant_Name":"new_plant"}
        )
    ]
)

Download Changes

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.

replica_sync_download_response = replica_based_kg.synchronize_replica(
    SynchronizeReplicaRequest(
        replica_id=new_replica_id, 
        download_parameters=SyncDownloadParameters(
            replica_deltas=ReplicaDeltas(
                last_sync_date=new_replica_sync_date.timestamp()*1000
            )
        )
    )
)

Upload Changes

The replica data in the download response can then be uploaded to the parent Knowledge Graph Service that the original replica was created on.

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.

For the individual steps that are part of _update_from_graph_replica() see the end of this guide.

knowledge_graph._update_from_graph_replica(create_result.replica_id,replica_sync_download_response.sync_data)

Check Service Updates

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.

result = knowledge_graph.query_streaming("MATCH (p:Plant) WHERE p.Plant_Name CONTAINS 'new_plant' RETURN p", as_dict=False)
list(result)

Steps to complete `_create_graph_from_replica()`

Note: These steps are written to be compatible with hosted knowledge graph services, you may need to make changes to handle your specific cases if you are using a service that is based on a NoSQL data store.

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:

Create a new knowledge graph service, this is where the replica data will be added

# set any additional features that should be supported
create_props = {"supportsProvenance": True, "supportsEditorTracking": False, "supportsSync": True}

# create the new service and connect to it
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})
replica_based_kg = KnowledgeGraph(new_kg_item.url, gis=gis)

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.

System maintained properties will be populated by the server and need to be dropped from the types in order to add them without errors.

service_dm = create_result.replica_data.service_data_model
all_system_maintained_properties = {} # collect system maintained properties, these will be used during data editing as well

for ent_type in service_dm.entity_types:
    system_maintained_properties = []
    # remove any types that are not regular
    if (ent_type.role != "esriGraphNamedObjectRegular"):
        service_dm.entity_types.remove(ent_type)
        continue
    props_to_remove = []
    # remove system maintained properties
    for prop in ent_type.properties:
        if prop.is_system_maintained == True:
            system_maintained_properties.append(prop.name)
            props_to_remove.append(ent_type.properties.index(prop))
    for prop_idx in sorted(props_to_remove, reverse=True):
        del ent_type.properties[prop_idx]
    indexes_to_remove = []
    # remove system maintained indexes
    for index in ent_type.field_indexes:
        if 'esri__' in index.name.lower():
            indexes_to_remove.append(ent_type.field_indexes.index(index))
    for index_idx in sorted(indexes_to_remove, reverse=True):
        del ent_type.field_indexes[index_idx]
    # add all system maintained properties to the dictionary for editing
    all_system_maintained_properties[ent_type.name] = system_maintained_properties
for rel_type in service_dm.relationship_types:
    system_maintained_properties = []
    # remove any types that are not regular
    if (rel_type.role != "esriGraphNamedObjectRegular"):
        service_dm.relationship_types.remove(rel_type)
        continue
    props_to_remove = []
    # remove system maintained properties
    for prop in rel_type.properties:
        if prop.is_system_maintained == True:
            system_maintained_properties.append(prop.name)
            #system_maintained_properties.remove('globalid')
            props_to_remove.append(rel_type.properties.index(prop))
    for prop_idx in sorted(props_to_remove, reverse=True):
        del rel_type.properties[prop_idx]
    indexes_to_remove = []
    # remove system maintained indexes
    for index in rel_type.field_indexes:
        if 'esri__' in index.name.lower():
            indexes_to_remove.append(rel_type.field_indexes.index(index))
    for index_idx in sorted(indexes_to_remove, reverse=True):
        del rel_type.field_indexes[index_idx]
    # add all system maintained properties to the dictionary for editing
    all_system_maintained_properties[rel_type.name] = system_maintained_properties
# add all of the entity and relationship types now that they are prepared
types_add = replica_based_kg.named_object_type_adds(entity_types=service_dm.entity_types, relationship_types=service_dm.relationship_types, as_dict=False)
# print the result to check that it was successful
print(types_add)

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())

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.

entity_adds = []
relationship_adds = []
# get the replica data frames stream
generator = create_result.replica_data.stream_data_frames()
while True:
    try:
        replica_data_frame = next(generator)
        # only use adds
        if (replica_data_frame.edit_type == "TYPE_ADD"):
            # iterate through each data frame
            items = replica_data_frame.iterate_data()
            for item in items:
                # get properties to delete based on system-maintained status and delete them from each instance
                delete_me = []
                for prop in item[0].properties.keys():
                    if prop in all_system_maintained_properties[item[0].type_name]:
                        delete_me.append(prop)
                for i in delete_me:
                    del item[0].properties[i]
                # create separate entity and relationship adds lists
                if type(item[0]) == Entity:
                    entity_adds.append(item[0])
                if type(item[0]) == Relationship:
                    relationship_adds.append(item[0])
    except StopIteration:
        break
# add entities first so they are there for the relationships
entity_response = replica_based_kg.apply_edits(adds=entity_adds, as_dict=False)
print(entity_response)
# add relationships which will attach to the created entities
relationship_response = replica_based_kg.apply_edits(adds=relationship_adds, as_dict=False)
print(relationship_response)

Set up and create a replica that contains all types so we can get changes made when we are ready to synchronize.

replicakg_replica_filters = ReplicaFilters(
    static_definition=StaticDefinition(
        entity_type_replica_definitions=[
            EntityTypeReplicaDefinition(type_name=entity_type.name)
            for entity_type in service_dm.entity_types
        ],
        relationship_type_replica_definitions=[
            RelationshipTypeReplicaDefinition(type_name=relationship_type.name)
            for relationship_type in service_dm.relationship_types
        ],
    )
)

# create replica using a replica request
replica_create_result = replica_based_kg.create_replica(
    ReplicaRequest(
        replica_definition = replicakg_replica_filters,
        replica_name = "replica_fulltypes_bidirectional",
        direction = "BIDIRECTIONAL"
    )
)

new_replica_id = replica_create_result.replica_id
new_replica_sync_date = replica_create_result.sync_date

Steps to complete `_update_from_graph_replica()`

The steps completed in that beta function can also be done manually, they are:

Create the list of edits for each edit type from the downloaded sync data

# create adds, updates, deletes from replica data
def generate_edit_types_lists(sync_data):
    adds = []
    updates = []
    deletes = []
    df_generator = sync_data.stream_data_frames()
    for replica_df in df_generator:
        print(replica_df.edit_type)
        if replica_df.edit_type == "TYPE_ADD":
            for edit in replica_df.iterate_data():
                adds.append(edit[0])
        if replica_df.edit_type == "TYPE_UPDATE":
            for edit in replica_df.iterate_data():
                updates.append(edit[0])
        if replica_df.edit_type == "TYPE_DELETE":
            for edit in replica_df.iterate_data():
                deletes.append(edit[0])
    return {"adds": adds, "updates": updates, "deletes": deletes}

Use those edits for synchronize_replica() uploads to the knowledge graph service.

all_edits = knowledge_graph.generate_edit_types_lists(replica_sync_download_response.sync_data)
knowledge_graph.synchronize_replica(
    SynchronizeReplicaRequest(
        replica_id=create_result.replica_id,
        upload_edits=SyncUploadEdits(
            sync_apply_edits=SyncApplyEdits(
                adds=all_edits["adds"],
                updates=all_edits["updates"],
                deletes=all_edits["deletes"]
            )
        )
    )
)