Showing posts with label HDFS. Show all posts
Showing posts with label HDFS. Show all posts

Monday, March 2, 2015

Node.js and Big Data


Node.js is a platform that is built on top of Chrome’s V8 javascript runtime which is ideal for building high performance, scalable and concurrent applications. The main feature of node.js is that it provides a evented, non-blocking I/O with a event loop which makes all IO operations asynchronous. 

node.js architecture



There are some NPM modules available for integrating with the Hadoop ecosystem. The table here shows the modules available for different components like HBase, HDFS, Hive, Solr and Zookeeper. Most of the modules is built on top of the Thrift or REST gateway exposed within each component. 

Node.js Hadoop Big Data Modules
Node.js is a perfect fit for I/O bound processes capable of handling a lot of concurrent requests. I have used these modules (highlighted in orange in the table above) in several use cases.

Tuesday, February 17, 2015

Configuring Cloudera Search to use local directory indexes

If you are using Cloudera Search, the SOLR collections that are created using Solrctl commands create collections that are backed in HDFS (org.apache.solr.core.HdfsDirectoryFactory) by default. If you want certain collections to use local indexes instead, then you can do the following:

1. Log in to one of the SOLR nodes and generate a new instance directory locally using the solrctl command:


solrctl instancedir —generate collectionConfigDir

The above command will generate the folder 'collectionConfigDir' with default configuration
2. Open the solrconfig.xml in the editor and make the following changes:

  • Comment out the section related to <directoryFactory> - The default generated config uses the HDFSDirectoryFactory. We can add a new section for the directoryFactory to use solr.NRTCachingDirectoryFactory


<directoryFactory name="DirectoryFactory" class="solr.NRTCachingDirectoryFactory">
</directoryFactory>

<!--  <directoryFactory name="DirectoryFactory" class="org.apache.solr.core.HdfsDirectoryFactory">
    <str name="solr.hdfs.home">${solr.hdfs.home:}</str>
    <str name="solr.hdfs.confdir">${solr.hdfs.confdir:}</str>
    <str name="solr.hdfs.security.kerberos.enabled">${solr.hdfs.security.kerberos.enabled:false}</str>
    <str name="solr.hdfs.security.kerberos.keytabfile">${solr.hdfs.security.kerberos.keytabfile:}</str>
    <str name="solr.hdfs.security.kerberos.principal">${solr.hdfs.security.kerberos.principal:}</str>
    <bool name="solr.hdfs.blockcache.enabled">${solr.hdfs.blockcache.enabled:true}</bool>
    <int name="solr.hdfs.blockcache.slab.count">${solr.hdfs.blockcache.slab.count:1}</int>
    <bool name="solr.hdfs.blockcache.direct.memory.allocation">${solr.hdfs.blockcache.direct.memory.allocation:true}</bool>
    <bool name="solr.hdfs.blockcache.direct.memory.allocation">false</bool>
    <int name="solr.hdfs.blockcache.blocksperbank">${solr.hdfs.blockcache.blocksperbank:16384}</int> 
    <bool name="solr.hdfs.blockcache.read.enabled">${solr.hdfs.blockcache.read.enabled:true}</bool>
    <bool name="solr.hdfs.blockcache.write.enabled">${solr.hdfs.blockcache.write.enabled:true}</bool>
    <bool name="solr.hdfs.nrtcachingdirectory.enable">${solr.hdfs.nrtcachingdirectory.enable:true}</bool>
    <int name="solr.hdfs.nrtcachingdirectory.maxmergesizemb">${solr.hdfs.nrtcachingdirectory.maxmergesizemb:16}</int>
    <int name="solr.hdfs.nrtcachingdirectory.maxcachedmb">${solr.hdfs.nrtcachingdirectory.maxcachedmb:192}</int>
  </directoryFactory> 
-->
  • Specify the solr data directory to use a local directory on the server. Also, ensure that the folder exists with required write permissions for the user running solr.

<!-- <dataDir>${solr.data.dir:}</dataDir>-->
<dataDir>/solr</dataDir>

  •  Change the lock type to use 'simple' instead of 'hdfs', which is set using this property 'solr.lock.type'

      <!--<lockType>${solr.lock.type:hdfs}</lockType>-->
 <lockType>simple</lockType>
3. Once these changes are done, then the instance directory can be uploaded as a new configuration to zookeeper using the following command

solrctrl instancedir --create collectionConfig collectionConfigDir

4. Now, the new collection can be created which will start using the local directory for storing the indexes.

solrctl collection --create newCollection -c collectionConfig -s 1 -r 2

The above command will create a collection (with 1 shard and 2 replicas) that will use local directory (/solr) for storing the indexes

Sunday, June 15, 2014

Performance Tuning and Optimization for Cloudera Search


In this post, I will explain some of the key factors that should be considered while optimizing performance (for querying and indexing) on implementations using Cloudera Search. Since Cloudera Search is built on top of SolrCloud, most of the performance considerations that are applicable for SOLR applications are still applicable for Cloudera Search. Please refer to this link for an overview of the main factors that affect SOLR performance (in general).

Below are some of the main factors that I found useful while optimizing performance on Cloudera Search

Block Caching
As we know, Cloudera Search uses HDFS filesystem to store the indexes. In order to optimize the performance, HDFS Block cache option is available. The block cahce works by caching the HDFS index blocks in JVM direct memory. Block cache can be enabled in solrconfig.xml. Below are the key parameters involved in tuning the block cache:

1. Enable Block Cache - by setting solr.hdfs.blockcache.enabled  to be true. Once block cache is enabled, the read and write caches can be enabled/disabled separately through the following settings:
  • solr.hdfs.blockcache.read.enabled
  • solr.hdfs.blockcache.write.enabled
 There is a known issue with enabling block cache writing which may lead to irrevocable corrupt indexes. So, it is very important that this is disabled by setting solr.hdfs.blockcache.read.enabled to be false.

2. Configure Memory Slab Count - The slab count determines the number of memory slabs to allocate, where each slab is 128 MB. Allocate the slab count to a sufficiently higher number that is required for the specific application (based on schema and query access patterns)
This is done by setting the solr.hdfs.blockcache.slab.count parameter

3.  Enable Global Block Cache - Enabling the global block cache would allow multiple solrcores on the same node to share a global HDFS block cache. It is done by setting solr.hdfs.blockcache.global to be true

4. NRTCachingDirectory - If using the Near Real-Time (NRT) setup, then enabling the NRTCachingDirectory (solr.hdfs.nrtcachingdirectory.enable) and tuning the maxCachedmb (solr.hdfs.nrtcachingdirectory.maxcachedmb) and merge size (solr.hdfs.nrtcachingdirectory.maxmergesizemb) thresholds also help.

Below is the xml section (in solrconfig.xml) with the above parameters:


<directoryFactory name="DirectoryFactory" class="solr.HdfsDirectoryFactory">
  <bool name="solr.hdfs.blockcache.enabled">true</bool>
  <bool name="solr.hdfs.blockcache.read.enabled">true</bool> 
  <bool name="solr.hdfs.blockcache.write.enabled">false</bool>
  <int name="solr.hdfs.blockcache.slab.count">100</int>
  <bool name="solr.hdfs.blockcache.direct.memory.allocation">true</bool>
  <int name="solr.hdfs.blockcache.blocksperbank">16384</int>
  <bool name="solr.hdfs.nrtcachingdirectory.enable">true</bool>
  <int name="solr.hdfs.nrtcachingdirectory.maxmergesizemb">64</int>
  <int name="solr.hdfs.nrtcachingdirectory.maxcachedmb">1024</int>
</directoryFactory>


SolrCloud Caching
The caching related settings play a major role in influencing the query performance/response time. Since SOLR caching are based on the Index Searches, the caches are available as long as the searchers are valid. Solr has support for FastLRUCache which has faster reads which can be set via (Solr.FastLRUCache) in  solrconfig.xml file. There are 3 types of caching available in SOLR

FilterCache - This cache stores unordered sets of document IDs matching the queries. It stores the results of any filter queries (fq) by executing and caching each filter separately. Setting this cache to sufficiently higher value helps in caching commonly used filter queries. Below is a sample configuration for the filter cache:

QueryResultCache - This cache stores the top N results of a query. Since it stores only the document IDs returned by the query, the memory usage of this cache is less compared to that of the filterCache.

DocumentCache - This cache stores the Lucene Document objects that have been fetched from disk. The size of the documentCache memory is dependent on the number and type of fields stored in the document.

Below is a sample configuration setting for the above three cache types:

<filterCache class="solr.FastLRUCache"
                 size="20000"
                 initialSize="5000"
                 autowarmCount="1000"/>

<queryResultCache class="solr.FastLRUCache"
                     size="10000"
                     initialSize="5000"
                     autowarmCount="1000"/>

<documentCache class="solr.FastLRUCache"
                   size="5000"
                   initialSize="2000"
                   autowarmCount="1000"/>

Disable Swapping
As with other hadoop systems, it is recommended to disable Linux swapping on all solr nodes.

To get more details on tuning Cloudera Search, please refer to this link

Friday, June 13, 2014

Handling OP_READ_BLOCK and File Not Found error in Cloudera Search

While using Cloudera Search in a (Near Real-Time) NRT scenario, occasionally during periods of high writes (index updates), I see the following error thrown on the solr logs

Failed to connect to /0.0.0.0:50010 for file /solr/collectionName/core_node1/data/index.20140528010609412/_zsae_Lucene41_0.doc for block BP-658256793-10.130.36.84-1390532185717:blk_-197867398637806450_45226458:java.io.IOException: Got error for OP_READ_BLOCK,​ self=/0.0.0.0:42978,​ remote=/0.0.0.0:50010,​ for file /solr/collectionName/core_node1/data/index.20140528010609412/_zsae_Lucene41_0.doc,​ for pool BP-658256793-10.130.36.84-1390532185717 block -197867398637806450_45226458

DFS chooseDataNode: got # 1 IOException,​ will wait for 1064.4631017325812 msec.

java.io.FileNotFoundException: File does not exist: /solr/collectionName/core_node1/data/index.20140528010609412/_zsae_Lucene41_0.doc

On troubleshooting further, the logs on datanode side show the following READ_BLOCK errors
DatanodeRegistration(<serverip>, storageID=DS-1520167466-0.0.0.0-50010-1389140294030, infoPort=50075, ipcPort=50020, storageInfo=lv=-40;cid=cluster25;nsid=1124238169;c=0):Got exception while serving BP-1661432518-0.0.0.0-1389140286721:blk_5701049520037157281_7691487 to /ip:45293
org.apache.hadoop.hdfs.server.datanode.ReplicaNotFoundException: Replica not found for BP-1661432518-10.130.71.219-1389140286721:blk_5701049520037157281_7691487
 at org.apache.hadoop.hdfs.server.datanode.BlockSender.getReplica(BlockSender.java:382)
 at org.apache.hadoop.hdfs.server.datanode.BlockSender.<init>(BlockSender.java:193)
 at org.apache.hadoop.hdfs.server.datanode.DataXceiver.readBlock(DataXceiver.java:326)
 at org.apache.hadoop.hdfs.protocol.datatransfer.Receiver.opReadBlock(Receiver.java:92)
 at org.apache.hadoop.hdfs.protocol.datatransfer.Receiver.processOp(Receiver.java:64)
 at org.apache.hadoop.hdfs.server.datanode.DataXceiver.run(DataXceiver.java:221)
 at java.lang.Thread.run(Thread.java:662)
node1:50010:DataXceiver error processing READ_BLOCK operation  src: /0.0.0.0:45293 dest: /0.0.0.0:50010
org.apache.hadoop.hdfs.server.datanode.ReplicaNotFoundException: Replica not found for BP-1661432518-0.0.0.0-1389140286721:blk_5701049520037157281_7691487
 at org.apache.hadoop.hdfs.server.datanode.BlockSender.getReplica(BlockSender.java:382)
 at org.apache.hadoop.hdfs.server.datanode.BlockSender.<init>(BlockSender.java:193)
 at org.apache.hadoop.hdfs.server.datanode.DataXceiver.readBlock(DataXceiver.java:326)
 at org.apache.hadoop.hdfs.protocol.datatransfer.Receiver.opReadBlock(Receiver.java:92)
 at org.apache.hadoop.hdfs.protocol.datatransfer.Receiver.processOp(Receiver.java:64)
 at org.apache.hadoop.hdfs.server.datanode.DataXceiver.run(DataXceiver.java:221)
 at java.lang.Thread.run(Thread.java:662)
This issue is related to this bug (https://issues.apache.org/jira/browse/SOLR-5693), where HDFS file merging does not work correctly with NRT search. This would lead to search failure with file not found exceptions. Again occurrence of the error depends on the velocity of the data changes. The issue has been backported to Cloudera Search and is supposed to be fixed with the next release of Search.

In versions 1.2 and below, a workaround to minimize the occurence of this failure is to do more frequent commits (reduce the autocommit frequency)

In the event of a catastrophic failure, when the replica goes down because of this issue and does NOT recover, you might have to manually copy over the index files on HDFS from another replica which is up and running. The error log specifies the actual index files that are missing because of the merge.

The index data files are usually stored in HDFS under /solr/{collection-name}/{core_node#}/data

There are two files in this directory which are important in identifying the state of the index:

  • index.properties - this file shows the current active index folder something like this:

#index.properties
#Wed Apr 30 23:57:38 PDT 2014
index=index.20140430235104131

  • replication.properties - this file shows replication state of the core

#Replication details
#Fri May 02 15:57:16 PDT 2014
previousCycleTimeInSeconds=486
indexReplicatedAtList=1399071436498,1399052241128,1398927458820,1398833578758,1398833453263,1398799104838
indexReplicatedAt=1399071436498
timesIndexReplicated=22
lastCycleBytesDownloaded=6012236791

I have used the "File Browser" application inside "Hue" to browse the files to get a  sense of where the files are located and which ones are missing. Below are the steps to fix the issue:

  1. Identify the core (eg: core_node1) and the file(s) missing from the logs
  2. Stop the solr process on the node with the failure. You can use Cloudera Manager to do this.
  3. Copy the index folder (index.xxxxx)from the replica that is working to the failed node. The copy can be done by either:

  • Using "hadoop fs -cp" command
  • Using Hue application.
     4. Copy the index.properties and replication.properties file.
     5. Restart the service.

In both cases, the copy operation has to be done in the context of a user (eg: hdfs) which has write permissions on the HDFS location.