Cassandra or Hadoop on a Just A Bunch Of Disks setup

I have been told (and implemented) that Hadoop nodes needn't be installed with RAID on, because that is overkill given the inherent replication of data in Hadoop; this also goes for Cassandra ( and i am assuming for other noSQL db's as well).

This is only assuming that as your data grows, you will be adding brand new nodes to your cluster / ring. This said, that means adding a brand new machine (VM or physical) with disks, RAM, ethernet connection, etc . This may not be always possible for businesses; in contrast, talking to a small startup recently, they had a data growth problem but didn't have the money to shell out more instances on their setup, so they just added more SATA disks to their nodes. 

Well, this is not a "planned" situation in Hadoop. When rebalancing the nodes, they found out that data was not being pushed out to these new disks .. 

So instead they had to reshape their cluster to stripe their disks via RAID - this allowed them to be able to add these new disks whenever they needed, without having to add new machines. Whenever a new disk needed to be added (or for that master if one disk failed), the node could just be taken out of the cluster, and hot swap a new disk was just a matter of minutes.

So this seems to be an argument against using a JBOD configuration, at least in Hadoop.

In Cassandra 1.2 it seems like extra-care has been added to utilize a JBOD configuration by default, but I believe the problem may be the same if you want to just add extra disks to your nodes..
See this link that describes this exact problem on Cassandra:
http://www.datastax.com/support-forums/topic/cassandra-on-jbod

Read More

How to perform ingestion and querying in a NOSQL Enterprise-level environment

I will talk about what I have seen in recent projects (i.e. this may not apply 2 years from now..). There is often the need for a replacement of a pure Oracle instance (expensive, scale-up architecture) to a NoSQL solution (much cheaper, scales out). The problem is replacing the functionality underneath to appropriately: - query the data. - ingest the data.

 Tools

 Datastax seems to be the strongest contender at the moment (Jul, 2013), by offering NoSQL, Hadoop and Solr functionality. Cloudera and MapR are following closely also with search functionality embedded. Both Datastax and Cloudera search have tight integration with HDFS and takes care of replication and sharding transparently by using the pre-existing hdfs replication and sharding, and use SolrCloud for this. However in the case of Cloudera search you would need to install zookeeper to enable coordination between each of the solr nodes; Zookeeper being standard in an HBase installation. DataStax allows you to talk to Solr, however their model scales around the data model and architecture of Cassandra. Hadoop is also available, as a convenience (and a secondary-tier performance compared to Cloudera/MapR).

  Querying

 So coming back to our problem, it seems like Solr is more or less replacing SQL in these types of architecture. Secondary indexes in Hbase or Cassandra unfortunately are seldomly used when you have a fair number of columns, with unique data. Solr allows you to do a fair amount of querying, up to simple aggregations via its Stats package. Customizations of the Stats handler are also pretty common. Unfortunately more complex queries in Solr, like GROUP BY HAVING clauses, start to break down the model and are not feasible.

  Ingesting

 You would think that Cassandra being so fast at writing would be best in a Datastax environment to perform the loading/ingesting of the data. Unfortunately after some extensive testing, it appears not to be the case, and you're much better of loading the data via Solr, using SolrJ, than using the popular Cassandra clients (Hector, Astyanax, Thrift), which prove too slow as an API for ingestion.

So Solr seems to be your best bet for this.
There are a number of gotchas to be aware of:

-Use Trye datatypes as opposed to 'normal' ones
-SolrJ's poorly documented API: how to get more than 10 docs returned by default, how to set the different querying options, etc
-A lot of 'search-centric' options of Solr don't really apply when you re just looking for a replacement of SQL. Things that come to mind: use field queries instead of queries (to not use the Heap), faceting.
-Solr lets you create a nice replacement to populating drop-downs, when 1000's of choices are available, with its autocomplete type of feature (a la Google).

Read More

Cassandra CQL-3 with wide rows / dynamic columns

So in the Cassandra community, people are not very happy with the move of Cassandra from its cli to CQL . Apparently 90% of existing Cassandra customers still use the cli/Thrift for their queries, and thus not very happy to have to move away from it.
One of the reasons for moving to CQL is, aside from obvious reasons like performance improvements and asynchronous calls, to compete with Mongo DB's simplicity actually.. But anyway, on to our subject of the day: my recent problem was that i wanted to insert at run time new columns on the fly in Cassandra (one of the primary reasons for using a NoSQL DB).

These new columns would come in with a Name, and a Value. So Datastax states that this is possible (http://www.datastax.com/dev/blog/does-cql-support-dynamic-columns-wide-rows ) in CQL ; however they dont state how to insert a column value along with the column name .

 Here is how:

 CREATE TABLE WideRow ( username VARCHAR, productNumber UUID, productName VARCHAR, purchaseDate TIMESTAMP, PRIMARY KEY (username, productName, productNumber) ;

 Essentially, create a composite column name with both the Name and the Value.
 These will be prepended to each of the values of the remaining (non-composite) columns. I.e., username is the row key. The other 2 columns create a composite column name, meaning they will be part of the column names, prepended to the other columns. I.e. 'matt', 'radio'_'123'_'04/01/2013', 'tv'_'235'_'05/02/2012' . This is a wide row, with 2 columns that each have different values. These 2 columns were created unique by using the Name and Value as a concatenation.

Read More

My presentation about how to integrate Apache Kafka, Camus, and Hive

Kafka camus presentation from mattlieber

Read More

My Note on Solutions.: Cassandra DSE, Testing Solr integration

My Note on Solutions.: Cassandra DSE, Testing Solr integration

everything worked for me also, except i got 
an error on the Solr search UI when querying saying"HTTP Status 500 - Unavailible shards"
 i had to do a 

'update keyspace WITH placement_strategy = 'NetworkTopologyStrategy' and strategy_options=[{Solr:1}]; ' 

to make things work ..

Anybody knows how to index data directly in Solr that is *not* first entered/created in Cassandra ? 
It seems like it is possible : " If you HTTP post the files to a non-existing column keyspace or column family, DSE Search creates the keyspace and column family, and then starts indexing the data. F"
from http://www.datastax.com/docs/datastax_enterprise2.0/search/dse_search_about

Read More

Using secondary indexes in Cassandra CQL: need to use an indexed column in your query

Say, i create this table:

cqlsh:Keyspace2> CREATE TABLE users (   user_name varchar PRIMARY KEY,   password varchar,   gender varchar,   session_token varchar,   state varchar,   birth_year bigint );

cqlsh:Keyspace2> INSERT INTO users

             ...          (user_name, password)

             ...          VALUES ('jsmith', 'ch@ngem3a');

cqlsh:Keyspace2> INSERT INTO users          (user_name, password)          VALUES ('jsmith2', 'ch@ngem3a2');

cqlsh:Keyspace2> create index on users (password);

I can create an index and query:

cqlsh:Keyspace2> select * from users where password = 'ch@ngem3a2' and user_name = 'jsmith2';

 user_name | birth_year | gender | password   | session_token | state

-----------+------------+--------+------------+---------------+-------

   jsmith2 |       1963 |   null | ch@ngem3a2 |          null |  null

 but this doesnt work:

cqlsh:Keyspace2> select * from users where birth_year > 1960;
Bad Request: No indexed columns present in by-columns clause with Equal operator
Perhaps you meant to use CQL 2? Try using the -2 option when starting cqlsh.
cqlsh:Keyspace2> select * from users where birth_year > 1960 and user_name 'jsmith';
Bad Request: line 1:48 no viable alternative at input 'user_name'
cqlsh:Keyspace2> select * from users where birth_year > 1960 and user_name = 'jsmith';
Bad Request: No indexed columns present in by-columns clause with Equal operator
Perhaps you meant to use CQL 2? Try using the -2 option when starting cqlsh.

Unless i use an indexed column in the query (not necessarily the primary indexed one):

cqlsh:Keyspace2> select * from users where birth_year > 1960  and password = 'ch@ngem3a2';
 user_name | birth_year | gender | password   | session_token | state
-----------+------------+--------+------------+---------------+-------
   jsmith2 |       1963 |   null | ch@ngem3a2 |          null |  null 

Read More

Trying Cassandra CQL

Here is how to create a new column on the fly in Cassandra's CQL2:

cqlsh:Keyspace2> select * from users;

 user_name | birth_year | gender | password   | session_token | state

-----------+------------+--------+------------+---------------+-------

   jsmith3 |       null |   null |        200 |          null |  null

    jsmith |       1968 |   null |  ch@ngem3a |          null |  null

   jsmith2 |       1963 |   null | ch@ngem3a2 |          null |  null

cqlsh:Keyspace1> insert into users (KEY, x) values ('jsmith',100);
cqlsh:Keyspace1> select * from users;
 KEY,TEST | birth_year,1968 | gender,m
 KEY,TEST1 | birth_year,1968 | gender,f
 KEY,jsmith | x,100

Read More

How to test Cassandra

Datastax's Cassandra comes with a stress tester, that can generate data for you.
It is all built in with different kind of parameters.
Here i am generating 1 M rows with 10 columns, readable, random values (up to 1000 values) that are indexed:

cassandra-stress -o INSERT  -n 1000000 -c 10 -U UTF8Type  -C 1000 --create-index=KEYS -r

Read More

How to use a Hive Avro Serde in distributed mode

Lately with our team, we tried to optimize our data by using an Avro Serde (with a binary encoding).
Unfortunately on CDH4, we would run into errors after creating the table;
Here is the Hive schema that we used:

Create external table avro_test 

ROW FORMAT SERDE

'org.apache.hadoop.hive.serde2.avro.AvroSerDe'

 STORED AS INPUTFORMAT

'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat' 

OUTPUTFORMAT

'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat' 

LOCATION

'/kafka/avro/topic_avro2/hourly/<date>/' TBLPROPERTIES ( 

'avro.schema.url'='file:///home/ubuntu/Message.avsc') ; 

But this didn't work :

hive>
    > select count(*) from avro_test;
Total MapReduce jobs = 1
Launching Job 1 out of 1
Number of reduce tasks determined at compile time: 1
In order to change the average load for a reducer (in bytes):
  set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
  set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
  set mapred.reduce.tasks=<number>
Starting Job = job_201303291126_0045, Tracking URL = ..Kill Command = /home/ubuntu/cdh4/hadoop-2.0.0-mr1-cdh4.2.0/bin/hadoop job  -kill job_201303291126_0045
Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 1
2013-03-29 20:50:47,871 Stage-1 map = 0%,  reduce = 0%
2013-03-29 20:51:22,151 Stage-1 map = 100%,  reduce = 100%
Ended Job = job_201303291126_0045 with errors
Error during job, obtaining debugging information...
Job Tracking URL: ..Examining task ID: task_201303291126_0045_m_000002 (and more) from job job_201303291126_0045
Task with the most failures(4):
-----
Task ID:
  task_201303291126_0045_m_000000
URL:..
-----
Diagnostic Messages for this Task:
java.io.IOException: java.lang.reflect.InvocationTargetExceptionat org.apache.hadoop.hive.io.HiveIOExceptionHandlerChain.handleRecordReaderCreationException(HiveIOExceptionHandlerChain.java:97)
at org.apache.hadoop.hive.io.HiveIOExceptionHandlerUtil.handleRecordReaderCreationException(HiveIOExceptionHandlerUtil.java:57)
at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileRecordReader.initNextRecordReader(HadoopShimsSecure.java:369)
at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileRecordReader.<init>(HadoopShimsSecure.java:316)
at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileInputFormatShim.getRecordReader(HadoopShimsSecure.java:430)
at org.apache.hadoop.hive.ql.io.CombineHiveInputFormat.getRecordReader(CombineHiveInputFormat.java:540)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:395)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:333)
at org.apache.hadoop.mapred.Child$4.run(Child.java:268)
at java.security.AccessController.doPrivileged(Native 

Well apparently the fix is to provide an inline schema instead:

CREATE EXTERNAL TABLE avro_topic

ROW FORMAT SERDE

'org.apache.hadoop.hive.serde2.avro.AvroSerDe'

STORED AS INPUTFORMAT

'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'

OUTPUTFORMAT

'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'

 LOCATION 

'/kafka/avro/topic_avro/hourly/2013/04/01/06'

TBLPROPERTIES (

'avro.schema.literal'='{

"namespace": "com.test.beans",

"type": "record",

"name": "Message",

"doc": "Logs for not so important stuff.",

"fields": [

{

"name": "id",

"type": "long",

"default":0

},

{

"name": "logTime",

"type": "long",

"default":0

},

{

"name": "muchoStuff",

"type": {"type": "map", "values": "string"},

"default":null

}

]

}');

Read More