Coherence Data Affinity Part 2 using KeyAssociator

In the earlier section we defined the association in the key object by implementing KeyAssociation.
The other way to do it is by writing a class that implements KeyAssociator and configuring that in cache-config.xml.


1. In this technique we will first create a key class. The difference between the earlier key and the current key is. The earlier one implements KeyAssociation but the current one doesn't.

package com.coher.dto;
import com.tangosol.io.pof.annotation.Portable;
import com.tangosol.io.pof.annotation.PortableProperty;
import java.io.Serializable;
@Portable
public class OrderLineItemKeyBean implements Serializable
{
  @PortableProperty
  private String mOrderNumber;
  @PortableProperty
  private String mLineItemNumber;
  public OrderLineItemKeyBean()
  {
    super();
  }
  public OrderLineItemKeyBean(String pOrderNumber, String pLineItemNumber)
  {
    super();
    this.mOrderNumber = pOrderNumber;
    this.mLineItemNumber = pLineItemNumber;
  }
  public void setOrderNumber(String pOrderNumber)
  {
    this.mOrderNumber = pOrderNumber;
  }
  public String getOrderNumber()
  {
    return mOrderNumber;
  }
  public void setLineItemNumber(String pLineItemNumber)
  {
    this.mLineItemNumber = pLineItemNumber;
  }
  public String getLineItemNumber()
  {
    return mLineItemNumber;
  }
  public String toString()
  {
    return mOrderNumber +" "+ mLineItemNumber;
  }
}
 
2. Lets now create a KeyAssociator class. This class actually implements KeyAssociator and is configured in the cache-config.xml. The getAssociatedKey method actually returns the associated key.


package com.coher.dataaffinity;
import com.coher.dto.OrderLineItemKeyBean;
import com.tangosol.net.PartitionedService;
import com.tangosol.net.partition.KeyAssociator;
public class MyKeyAssociator
  implements KeyAssociator
{
  public MyKeyAssociator()
  {
    super();
  }
  public void init(PartitionedService partitionedService)
  {
  }
  public Object getAssociatedKey(Object object)
  {
    if (object instanceof OrderLineItemKeyBean)
    {
      return ((OrderLineItemKeyBean) object).getOrderNumber();
    }
    else
    {
      return object;
    }
  }
}


3. Configure the KeyAssociator class in cache-config.xml

<?xml version="1.0" ?>
<cache-config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
              xmlns="http://xmlns.oracle.com/coherence/coherence-cache-config"
              xsi:schemaLocation="http://xmlns.oracle.com/coherence/coherence-cache-config coherence-cache-config.xsd">
<scope-name>OneSystemScope</scope-name>
<defaults>
  <serializer>pof</serializer>
</defaults>
<caching-scheme-mapping>
  <cache-mapping>
   <cache-name>mycache</cache-name>
   <scheme-name>dist-default</scheme-name>
   <init-params>
    <init-param>
     <param-name>size-limit</param-name>
     <param-value>5000</param-value>
    </init-param>
   </init-params>
  </cache-mapping>
  <cache-mapping>
   <cache-name>DistributedOrders</cache-name>
   <scheme-name>dist-default</scheme-name>
   <init-params>
    <init-param>
     <param-name>size-limit</param-name>
     <param-value>5000</param-value>
    </init-param>
   </init-params>
  </cache-mapping>
  <cache-mapping>
   <cache-name>DistributedOrderLines</cache-name>
   <scheme-name>dist-default</scheme-name>
   <init-params>
    <init-param>
     <param-name>size-limit</param-name>
     <param-value>5000</param-value>
    </init-param>
   </init-params>
  </cache-mapping>
</caching-scheme-mapping>
<caching-schemes>
  <distributed-scheme>
   <scheme-name>dist-default</scheme-name>
   <service-name>DistributedCache</service-name>
   <thread-count>5</thread-count>
   <backup-count>1</backup-count>
   <backup-storage>
    <type>file-mapped</type>
    <initial-size>1M</initial-size>
    <maximum-size>5G</maximum-size>
    <directory>/software/CoherenceInsall/coherence/backup</directory>
   </backup-storage>
   <key-associator>
    <class-name>com.coher.dataaffinity.MyKeyAssociator</class-name>
   </key-associator>   <backing-map-scheme>
    <overflow-scheme>
     <scheme-name>LocalMemoryWithDiskOverflow</scheme-name>
     <front-scheme>
      <local-scheme>
       <high-units>300000</high-units>
      </local-scheme>
     </front-scheme>
     <back-scheme>
      <external-scheme>
       <scheme-name>DiskScheme</scheme-name>
       <nio-file-manager>
        <initial-size>1MB</initial-size>
        <maximum-size>1024MB</maximum-size>
        <directory>/software/CoherenceInstall/coherence/backup</directory>
       </nio-file-manager>
      </external-scheme>
     </back-scheme>
    </overflow-scheme>
   </backing-map-scheme>
   <autostart>true</autostart>
  </distributed-scheme>
  <invocation-scheme>
   <scheme-name>InvocationService</scheme-name>
   <service-name>InvocationService</service-name>
   <thread-count>5</thread-count>
   <autostart>true</autostart>
  </invocation-scheme>
</caching-schemes>
</cache-config>
 
4. Add the key class to pof-config.xml. If you are using POF.
 
5. Create a cache loader class to test
 
package com.coher.dataaffinity;
import com.coher.dto.OrderBean;
import com.coher.dto.OrderLineItemBean;
import com.coher.dto.OrderLineItemKeyBean;
import com.tangosol.net.CacheFactory;
import com.tangosol.net.NamedCache;
import java.util.Random;
public class DataAffinityLoader
{
  public DataAffinityLoader()
  {
    super();
  }
  public static void main(String[] args)
  {
    NamedCache orderCache = CacheFactory.getCache("DistributedOrders");
    NamedCache orderLineCache =
      CacheFactory.getCache("DistributedOrderLines");
    orderCache.clear();
    orderLineCache.clear();
    int k = 0;
    for (int i = 0; i < 4; i++)
    {
      OrderBean ob = new OrderBean("" + i, "Customer" + i);
      orderCache.put(ob.getOrderNumber(), ob);
      Random r = new Random();
      int l = r.nextInt(20);
      for (int j = 0; j <= 3; j++)
      {
        k++;
        OrderLineItemBean olib =
          new OrderLineItemBean(ob.getOrderNumber(), "lineitem" + k,
                                "" + k, k);
        orderLineCache.put(new OrderLineItemKeyBean(ob.getOrderNumber(),
                                              olib.getItemNumber()), olib);
      }
    }
  }
}


6. Create a class to test the affinity

package com.coher.dataaffinity;
import com.coher.dto.OrderLineItemKeyBean;
import com.tangosol.net.CacheFactory;
import com.tangosol.net.Member;
import com.tangosol.net.NamedCache;
import com.tangosol.net.PartitionedService;
import com.tangosol.util.Filter;
import com.tangosol.util.filter.EqualsFilter;
import com.tangosol.util.filter.KeyAssociatedFilter;
import java.util.Set;
public class DataAffinityCacheViewer
{
  public DataAffinityCacheViewer()
  {
    super();
  }
  public static void main(String[] args)
  {
    NamedCache orderCache = CacheFactory.getCache("DistributedOrders");
    NamedCache orderLineCache = CacheFactory.getCache("DistributedOrderLines");
    System.out.println("orders Cache.size() = " + orderCache.size());
    PartitionedService ps1 = (PartitionedService) orderCache.getCacheService();
    Set<String> odrKeySet = (Set<String>) orderCache.keySet();
    for (String key: odrKeySet)
    {
      Member member = ps1.getKeyOwner(key);
      System.out.println("Coherence member:" + member.getId() + "; order key:" + key );
      EqualsFilter filterEq = new EqualsFilter("getOrderNumber", key);
      Filter filterAsc = new KeyAssociatedFilter(filterEq, key);
      Set<OrderLineItemKeyBean> ONKeySet = (Set<OrderLineItemKeyBean>) orderLineCache.keySet(filterAsc);
      PartitionedService ps2 = (PartitionedService) orderLineCache.getCacheService();
      for (OrderLineItemKeyBean key1: ONKeySet)
      {
        Member member1 = ps2.getKeyOwner(key1);
        System.out.println("              Coherence member:" + member1.getId() + "; Order Line Key:" + key1 );
      }
    } 
  }
}
 
Running the test.

a. Start two instances of coherence
b. first run DataAffinityLoader to load data
c. Now run DataAffinityCacheViewer to test affinity
orders Cache.size() = 4
Coherence member:3; order key:0
              Coherence member:3; Order Line Key:0 lineitem4
              Coherence member:3; Order Line Key:0 lineitem1
              Coherence member:3; Order Line Key:0 lineitem3
              Coherence member:3; Order Line Key:0 lineitem2
Coherence member:3; order key:1
              Coherence member:3; Order Line Key:1 lineitem7
              Coherence member:3; Order Line Key:1 lineitem8
              Coherence member:3; Order Line Key:1 lineitem6
              Coherence member:3; Order Line Key:1 lineitem5
Coherence member:1; order key:2
              Coherence member:1; Order Line Key:2 lineitem12
              Coherence member:1; Order Line Key:2 lineitem11
              Coherence member:1; Order Line Key:2 lineitem9
              Coherence member:1; Order Line Key:2 lineitem10
Coherence member:1; order key:3
              Coherence member:1; Order Line Key:3 lineitem14
              Coherence member:1; Order Line Key:3 lineitem15
              Coherence member:1; Order Line Key:3 lineitem16
              Coherence member:1; Order Line Key:3 lineitem13

Coherence Data Affinity Part 1 using KeyAssociation

Data affinity can be used to keep related objects in the same partition. If you have an order and line item caches, We can use this to keep related objects together thus increasing the efficiency of coherence.

The objects can be unevenly distributed if the child keys are not unique.

This can be done in two ways
1. Using a Key that implements KeyAssociation Interface in the Child Cache
     using this method is illustrated below

2. Creating a class that implements KeyAssociator and configuring it CacheConfig.xml



Below is an illustration of how to achieve it using KeyAssociation

1. Lets create a Order class
package com.coher.dto;
import com.tangosol.io.pof.annotation.Portable;
import com.tangosol.io.pof.annotation.PortableProperty;
import java.io.Serializable;
@Portable
public class OrderBean
  implements Serializable
{
  public OrderBean()
  {
    super();
  }
  @PortableProperty
  private String mOrderNumber;
  @PortableProperty
  private String mCustomerName;
  public OrderBean(String mOrderNumber, String mCustomerName)
  {
    super();
    this.mOrderNumber = mOrderNumber;
    this.mCustomerName = mCustomerName;
  }
  public void setOrderNumber(String mOrderNumber)
  {
    this.mOrderNumber = mOrderNumber;
  }
  public String getOrderNumber()
  {
    return mOrderNumber;
  }
  public void setCustomerName(String mCustomerName)
  {
    this.mCustomerName = mCustomerName;
  }
  public String getCustomerName()
  {
    return mCustomerName;
  }
}

2. Lets Create a line item class
package com.coher.dto;
import com.tangosol.io.pof.annotation.Portable;
import com.tangosol.io.pof.annotation.PortableProperty;
import java.io.Serializable;
@Portable
public class OrderLineItemBean
  implements Serializable
{
  public OrderLineItemBean()
  {
    super();
  }
  @PortableProperty
  String mLineItemNumber;
  @PortableProperty
  String mOrderNumber;
  @PortableProperty
  String mItemName;
  @PortableProperty
  Integer mQuantity;
  public OrderLineItemBean(String pOrderNumber, String pLineItemNumber,
                           String pItemName, Integer pQuantity)
  {
    super();
    this.mOrderNumber = pOrderNumber;
    this.mLineItemNumber = pLineItemNumber;
    this.mItemName = pItemName;
    this.mQuantity = pQuantity;
  }
  public void setOrderNumber(String pOrderNumber)
  {
    this.mOrderNumber = pOrderNumber;
  }
  public String getOrderNumber()
  {
    return mOrderNumber;
  }
  public void setItemNumber(String pLineItemNumber)
  {
    this.mLineItemNumber = pLineItemNumber;
  }
  public String getItemNumber()
  {
    return mLineItemNumber;
  }
  public void setItemName(String pItemName)
  {
    this.mItemName = pItemName;
  }
  public String getItemName()
  {
    return mItemName;
  }
  public void setQuantity(Integer pQuantity)
  {
    this.mQuantity = pQuantity;
  }
  public Integer getQuantity()
  {
    return mQuantity;
  }
  public String toString()
  {
    return mLineItemNumber + " - " + mOrderNumber + " - " + mItemName +
      " - " + mQuantity;
  }
}

3. Create a key association.
This class is used as a key for the detail cache entries and

package com.coher.dto;
import com.tangosol.io.pof.annotation.Portable;
import com.tangosol.io.pof.annotation.PortableProperty;
import com.tangosol.net.cache.KeyAssociation;
import java.io.Serializable;
@Portable
public class OrderNumberKey implements KeyAssociation, Serializable
{
  @PortableProperty
  private String mOrderNumber;
  @PortableProperty
  private String mLineItemNumber;
  public OrderNumberKey()
  {
    super();
  }
  public OrderNumberKey(String pOrderNumber, String pLineItemNumber)
  {
    super();
    this.mOrderNumber = pOrderNumber;
    this.mLineItemNumber = pLineItemNumber;
  }
  public Object getAssociatedKey()
  {
    return getOrderNumber();
  }
  public void setOrderNumber(String pOrderNumber)
  {
    this.mOrderNumber = pOrderNumber;
  }
  public String getOrderNumber()
  {
    return mOrderNumber;
  }
  public void setLineItemNumber(String pLineItemNumber)
  {
    this.mLineItemNumber = pLineItemNumber;
  }
  public String getLineItemNumber()
  {
    return mLineItemNumber;
  }
  
  public String toString()
  {
    return mOrderNumber +" "+ mLineItemNumber;
  }
}

4. Package the jar and place it in the classpath of coherence server



5.  Lets create a class that loads the cache. In this cache when loading the detail objects into the cache we use the Key Association we created above in Step 3.

package com.coher.dataaffinity;
import com.coher.dto.OrderBean;
import com.coher.dto.OrderLineItemBean;
import com.coher.dto.OrderNumberKey;
import com.tangosol.net.CacheFactory;
import com.tangosol.net.NamedCache;
import java.util.Random;
public class DataAffinityLoader
{
  public DataAffinityLoader()
  {
    super();
  }
  public static void main(String[] args)
  {
    NamedCache orderCache = CacheFactory.getCache("DistributedOrders");
    NamedCache orderLineCache =
      CacheFactory.getCache("DistributedOrderLines");
    orderCache.clear();
    orderLineCache.clear();
    int k = 0;
    for (int i = 0; i < 4; i++)
    {
      OrderBean ob = new OrderBean("" + i, "Customer" + i);
      orderCache.put(ob.getOrderNumber(), ob);
      Random r = new Random();
      int l = r.nextInt(20);
      for (int j = 0; j <= 3; j++)
      {
        k++;
        OrderLineItemBean olib =
          new OrderLineItemBean(ob.getOrderNumber(), "lineitem" + k,
                                "" + k, k);
        orderLineCache.put(new OrderNumberKey(ob.getOrderNumber(),
                                              olib.getItemNumber()), olib);
      }
    }
  }
}

6. Lets create a class that reads and shows us the distribution
package com.coher.dataaffinity;
import com.coher.dto.OrderNumberKey;
import com.tangosol.net.CacheFactory;
import com.tangosol.net.Member;
import com.tangosol.net.NamedCache;
import com.tangosol.net.PartitionedService;
import com.tangosol.util.Filter;
import com.tangosol.util.filter.EqualsFilter;
import com.tangosol.util.filter.KeyAssociatedFilter;
import java.util.Set;
public class DataAffinityCacheViewer
{
  public DataAffinityCacheViewer()
  {
    super();
  }
  public static void main(String[] args)
  {
    NamedCache orderCache = CacheFactory.getCache("DistributedOrders");
    NamedCache orderLineCache = CacheFactory.getCache("DistributedOrderLines");
    System.out.println("orders Cache.size() = " + orderCache.size());
    PartitionedService ps1 = (PartitionedService) orderCache.getCacheService();
    Set<String> odrKeySet = (Set<String>) orderCache.keySet();
    for (String key: odrKeySet)
    {
      Member member = ps1.getKeyOwner(key);
      System.out.println("Coherence member:" + member.getId() + "; order key:" + key );
      EqualsFilter filterEq = new EqualsFilter("getOrderNumber", key);
      Filter filterAsc = new KeyAssociatedFilter(filterEq, key);
      Set<OrderNumberKey> ONKeySet = (Set<OrderNumberKey>) orderLineCache.keySet(filterAsc);
      PartitionedService ps2 = (PartitionedService) orderLineCache.getCacheService();
      for (OrderNumberKey key1: ONKeySet)
      {
        Member member1 = ps2.getKeyOwner(key1);
        System.out.println("              Coherence member:" + member1.getId() + "; Order Line Key:" + key1 );
      }
    }
  }
}

7. Start the caches
8. Run the Data affinity loader to load the cache
9. Run the  cache viewer to  read the cache
Coherence member:1; order key:2
              Coherence member:1; Order Line Key:2 lineitem9
              Coherence member:1; Order Line Key:2 lineitem12
              Coherence member:1; Order Line Key:2 lineitem11
              Coherence member:1; Order Line Key:2 lineitem10
Coherence member:1; order key:3
              Coherence member:1; Order Line Key:3 lineitem14
              Coherence member:1; Order Line Key:3 lineitem15
              Coherence member:1; Order Line Key:3 lineitem13
              Coherence member:1; Order Line Key:3 lineitem16
Coherence member:5; order key:0
              Coherence member:5; Order Line Key:0 lineitem3
              Coherence member:5; Order Line Key:0 lineitem2
              Coherence member:5; Order Line Key:0 lineitem4
              Coherence member:5; Order Line Key:0 lineitem1
Coherence member:5; order key:1
              Coherence member:5; Order Line Key:1 lineitem5
              Coherence member:5; Order Line Key:1 lineitem8
              Coherence member:5; Order Line Key:1 lineitem7
              Coherence member:5; Order Line Key:1 lineitem6

rohith raj puchalapalli