Indexes

WSO2 DAS has a distributed indexing engine that is built on top of Apache Lucene. Data is indexed and saved in the Data Access Layer of DAS in a store referred to as the Analytics File System. This section explains how indexing is carried out in WSO2 DAS.

A main requirement of DAS is to support arbitrary key value pairs. Most of the RDBMS have a fixed schema and do not support storing arbitrary key values. In DAS, records are stored as blobs. This way, the data fields in a record can be converted to a blob and stored together with the record ID and the timestamp of the record. Storing data as blobs has one drawback. That is database level indexing cannot be used for blobs to search by fields. To overcome this issue, records are sent through the indexer before the records are persisted in DAS so that the indexer can keep a searchable copy of record data along with the record ID. This indexer does not store the field values. It only stores the record ID and keeps an index of data fields. When a search query is sent to DAS, the indexer identifies the matching records, gets their record IDs, and sends them to the persistence layer/recordstore. In recordstore level, the record blobs are deserialized and returned as the record objects from the AnalyticsDataService implementation.

Indexing Architecture

DAS indexer is implemented using Apache Lucene which is a full text search library. Users can index records and search for records later via Lucene queries. Events received by DAS are converted to a list of  records and inserted into FileSystem based queues. These queues are created in <DAS_HOME>/repository/data/index_staging_queues directory. With a background thread, these queues are consumed and records are indexed. The indexed data is stored in the <DAS_HOME>/repository/data/index_data directory. The DAS index consists of smaller indexes known as shards. A shard can be accessed nly by one index writer at a given time. Therefore having multiple shards can increase the write throughput (however, the write throughput can be limited by Disk IO operations). By default, DAS is configured to have 6 shards and 1 replica.

Indexing related configurations are done in the <DAS_HOME>/repository/conf/analytics/analytics-config.xml file. Additionally, information relating to shards is maintained in the <DAS_HOME>/repository/conf/analytics/local-shard-allocation-config.conf file. This file stores the shard number along with its state (that can be INIT or NORMAL). The INIT is the initial state. Usually this state cannot be seen from outside. This is because the INIT state changes to the NORMAL state once the server starts. If the indexing node is running, the state of shards should be NORMAL and not INIT. The NORMAL state denotes that the indexing node has started indexing. Therefore, whenever the data is ready to be indexed, the indexer node indexes the incoming data.

To re-index the whole dataset, follow the steps below:

1. Shut down the WSO2 DAS server.
2. Remove all the index data stored in the <DAS_HOME>/repository/data directory.
3. In the <DAS_HOME>/repository/conf/analytics/local-shard-allocation-config.conf file, change the mode for all the shards from NORMAL to INIT.
4. Restart the WSO2 DAS server.

By default, DAS is configured to have 6 shards and 1 replica for each shard(shards and replicas, altogether 12 shards). So in Min-HA setup (two node cluster), each DAS server will contain 3 shards and replicas of the other 3 shards. This is communicated between the nodes through Hazelcast messages. Even if one server goes down, second server will have the replicas of the 3 shards which were in the node that went down, so HA is preserved. For example,  If there are 3 DAS nodes in the cluster, all 12 shards (shards + replicas) will be split among the 3 nodes where each node will have 4 shards (2 shards +  any 2 replicas from other 4 shards). 

Debugging Indexing in a cluster

If someone thinks there is an issue with indexing, first thing to do is, check the local-shard-allocation.config.conf and see if the shards are allocated properly. The number of shards and the number of replicas are mentioned in analytics-config.xml.

<!-- The number of index data replicas the system should keep, for H/A, this should be at least 1, e.g. the value 0 means
        there aren't any copies of the data -->
   <indexReplicationFactor>1</indexReplicationFactor>
   <!-- The number of index shards, should be equal or higher to the number of indexing nodes that is going to be working,
        ideal count being 'number of indexing nodes * [CPU cores used for indexing per node]' -->
   <shardCount>6</shardCount>

Second thing to check is the logs that are printed when the cluster is initialized.

When  DAS starts, it will print a log line similar to below one.

TID: [-1] [] [2017-05-12 13:17:27,538]  INFO {org.wso2.carbon.analytics.dataservice.core.indexing.IndexNodeCoordinator} -  Indexing Initialized: CLUSTERED {0={c13d3a23-b15a-4b9c-ac8f-a30df2811c98=Member [10.3.24.70]:4000 this, bc751d36-d345-4b8f-b133-b77793f04805=Member [10.3.24.67]:4000}, 1={c13d3a23-b15a-4b9c-ac8f-a30df2811c98=Member [10.3.24.70]:4000 this, bc751d36-d345-4b8f-b133-b77793f04805=Member [10.3.24.67]:4000}, 2={c13d3a23-b15a-4b9c-ac8f-a30df2811c98=Member [10.3.24.70]:4000 this, bc751d36-d345-4b8f-b133-b77793f04805=Member [10.3.24.67]:4000}, 3={c13d3a23-b15a-4b9c-ac8f-a30df2811c98=Member [10.3.24.70]:4000 this, bc751d36-d345-4b8f-b133-b77793f04805=Member [10.3.24.67]:4000}, 4={c13d3a23-b15a-4b9c-ac8f-a30df2811c98=Member [10.3.24.70]:4000 this, bc751d36-d345-4b8f-b133-b77793f04805=Member [10.3.24.67]:4000}, 5={c13d3a23-b15a-4b9c-ac8f-a30df2811c98=Member [10.3.24.70]:4000 this, bc751d36-d345-4b8f-b133-b77793f04805=Member [10.3.24.67]:4000}} | Current Node Indexing: Yes {org.wso2.carbon.analytics.dataservice.core.indexing.IndexNodeCoordinator}

Here {0={c13d3a23-b15a-4b9c-ac8f-a30df2811c98=Member [10.36.241.70]:4000 this, bc751d36-d345-4b8f-b133-b77793f04805=Member [10.36.241.67]:4000} means that, the shard 0 is allocated to 2 nodes and their ids are c13d3a23-b15a-4b9c-ac8f-a30df2811c98 and bc751d36-d345-4b8f-b133-b77793f04805. Also their respective ips are mentioned in the log line. So, For correctly configured 2 node DAS cluster, this log line should contain all 6 shards ( from 0 to 5) and their respective node ids. When you initially setup the cluster and start the first server, this log line will not contain the 2 node ids, as the other node has not joined the cluster previously. When the second node joins the cluster, a similar log line will be printed with complete map of shard along with node ids. The two node ids mentioned in the log line should match with the ids mentioned in the my-node-id.dat in both DAS nodes. In a two node cluster, if there are more than 2 unique ids, shard allocation is messed and needs to be fixed. One reason for containing more than 2 ids is that, reconfiguring the cluster with new DAS packs without clearing the data sources in analytics-datasources.xml. When a new DAS pack is started pointing to the same analytics data sources, a new id is generated and the new DAS pack will be considered the third node who joined the cluster. When we are re-configuring the cluster with new DAS packs, we need to make sure that we first backup the older ids and restore them in the new packs. If the two node cluster already contain more than 2 unique ids ( due to starting a new DAS pack with new id), we have to remove the unnecessary/additional ids. Steps are as below.

1. Keep backup of the current my-node-id.dat of both nodes.

2. Extract the node ids mentioned in the log line mentioned above. This may contain more than 2 ids.

3. Separate out the ids which do not match with the my-node-id.dat of both nodes.

4. Shutdown both nodes.
5. Get one of the non-matching/unnecessary/additional node ids and replace the my-node-id.dat of any DAS node. (make sure you keep a backup of the my-node-id.dat before doing this) 

6. Start the DAS node mentioned in step 5 with property “-DdisableIndexing=true”. This command will remove the node id that we put in my-node-id.dat from the indexing cluster.

7. Repeat the steps 4, 5, and 6 for all the additional node ids.

8. Finally, restore the two node ids (restore the my-node-id.dat files that you backed up).

9. Clean the repository/data folder

10. Go to local-shard-allocation.config.conf and replace the content with following.
    0, INIT
    1, INIT
    2, INIT
    3, INIT
    4, INIT
    5, INIT

11. Do the above for both nodes and start. You will see that the local-shard-allocation-config.conf content is changed to

    0, NORMAL
    1, NORMAL
    2, NORMAL
    3, NORMAL
    4, NORMAL
    5, NORMAL

    Note that the above configuration is valid a DAS configured with 6 shards and 1 replicas. Step 10 should be changed according to the number of shards and replicas you have.

Indexing in Spark Environment

Spark runs in a different JVM. Therefore, Spark cannot write directly to Indexing queues. When we instruct the spark to index its summarized data, what Spark actually does is, it sends the summarized data to be indexed, to a storage area called staging area in the primary recordstore. So using the staging Index worker threads, these data is fetched and inserted to indexing queue. From that point onwards the indexing process is the same. Usually indexing can be slow if we frequently run the spark scripts (if those spark scripts contain instructions to index summarized data). Sometimes, there can be situations where the same data set is processed within a small time frame (e.g. spark scripts scheduled to run every 2 mins with indexing). In those cases, the same data is indexed over and over again. Best thing is to use incremental processing, So the reindexing of same data can be minimized.