Cassandra: The Definitive Guide

1. Simple Snitch
   
2. 
   
3. By default, Cassandra uses org.apache.cassandra.locator.EndPointSnitch. It operates by simply comparing different octets in the IP addresses of each node. If two hosts have the same value in the second octet of their IP addresses, then they are determined to be in the same data center. If two hosts have the same value in the third octet of their IP addresses, then they are determined to be in the same rack. “Determined to be” really means that Cassandra has to guess based on an assumption of how your servers are located in different VLANs or subnets.

复制代码

1. PropertyFileSnitch
   
2. 
   
3. The org.apache.cassandra.locator.PropertyFileSnitch used to be in contrib, but was moved into the main code base in 0.7. This snitch allows you more control when using a Rack-Aware Strategy by specifying node locations in a standard key/value properties file called cassandra-rack.properties.
   
4. 
   
5. This snitch was contributed by Digg, which uses Cassandra and regularly contributes to its development. This snitch helps Cassandra know for certain if two IPs are in the same data center or on the same rack—because you tell it that they are. This is perhaps most useful if you move servers a lot, as operations often need to, or if you have inherited an unwieldy IP scheme.
   
6. 
   
7. The default configuration of cassandra-rack.properties looks like this:
   
8. 
   
9. # Cassandra Node IP=Data Center:Rack
   
10. 10.0.0.10=DC1:RAC1
    
11. 10.0.0.11=DC1:RAC1
    
12. 10.0.0.12=DC1:RAC2
    
13. 
    
14. 10.20.114.10=DC2:RAC1
    
15. 10.20.114.11=DC2:RAC1
    
16. 10.20.114.15=DC2:RAC2
    
17. 
    
18. # default for unknown nodes
    
19. default=DC1:r1

复制代码

1. Basic Write Properties
   
2. 
   
3. There are a few basic properties of Cassandra’s write ability that are worth noting. First, writing data is very fast in Cassandra, because its design does not require performing disk reads or seeks. The memtables and SSTables save Cassandra from having to perform these operations on writes, which slow down many databases. All writes in Cassandra are append-only.
   
4. 
   
5. Because of the database commit log and hinted handoff design, the database is always writeable, and within a column family, writes are always atomic.

复制代码

1. Example 7-3. SlicePredicate.java
   
2. package com.cassandraguide.rw;
   
3. 
   
4. // imports omitted
   
5. 
   
6. public class SlicePredicateExample {
   
7.        
   
8.         public static void main(String[] args) throws Exception {
   
9.                 Connector conn = new Connector();
   
10.                 Cassandra.Client client = conn.connect();
    
11.                
    
12.                 SlicePredicate predicate = new SlicePredicate();
    
13.                 List<byte[]> colNames = new ArrayList<byte[]>();
    
14.                 colNames.add("a".getBytes());
    
15.                 colNames.add("b".getBytes());
    
16.                 predicate.column_names = colNames;
    
17.                
    
18.                 ColumnParent parent = new ColumnParent("Standard1");
    
19. 
    
20.                 byte[] key = "k1".getBytes();
    
21.                 List<ColumnOrSuperColumn> results =
    
22.                         client.get_slice(key, parent, predicate, ConsistencyLevel.ONE);
    
23.                
    
24.                 for (ColumnOrSuperColumn cosc : results) {       
    
25.                         Column c = cosc.column;
    
26.                         System.out.println(new String(c.name, "UTF-8") + " : "
    
27.                                         + new String(c.value, "UTF-8"));
    
28.                 }
    
29.                                
    
30.                 conn.close();
    
31.                
    
32.                 System.out.println("All done.");
    
33.         }
    
34. }

复制代码

1. Example 7-4. GetRangeSliceExample.java
   
2. package com.cassandraguide.rw;
   
3. 
   
4. //imports omitted
   
5. 
   
6. public class GetRangeSliceExample {
   
7.        
   
8.         public static void main(String[] args) throws Exception {
   
9.                 Connector conn = new Connector();
   
10.                 Cassandra.Client client = conn.connect();
    
11.                
    
12.                 System.out.println("Getting Range Slices.");
    
13.                
    
14.                 SlicePredicate predicate = new SlicePredicate();
    
15.                 List<byte[]> colNames = new ArrayList<byte[]>();
    
16.                 colNames.add("a".getBytes());
    
17.                 colNames.add("b".getBytes());
    
18.                 predicate.column_names = colNames;
    
19.                
    
20.                 ColumnParent parent = new ColumnParent("Standard1");
    
21.                
    
22.                 KeyRange keyRange = new KeyRange();
    
23.                 keyRange.start_key = "k1".getBytes();
    
24.                 keyRange.end_key = "k2".getBytes();
    
25. 
    
26.                 //a key slice is returned
    
27.                 List<KeySlice> results =
    
28.                         client.get_range_slices(parent, predicate, keyRange,
    
29.                                         ConsistencyLevel.ONE);
    
30.                
    
31.                 for (KeySlice keySlice : results) {       
    
32.                         List<ColumnOrSuperColumn> cosc = keySlice.getColumns();
    
33.                                                
    
34.                         System.out.println("Current row: " +
    
35.                                         new String(keySlice.getKey()));
    
36.                        
    
37.                         for (int i = 0; i < cosc.size(); i++) {       
    
38.                                 Column c = cosc.get(i).getColumn();
    
39.                                 System.out.println(new String(c.name, "UTF-8") + " : "
    
40.                                                 + new String(c.value, "UTF-8"));
    
41.                         }
    
42.                 }
    
43.                                
    
44.                 conn.close();
    
45.                
    
46.                 System.out.println("All done.");
    
47.         }
    
48. }

复制代码

1. Example 7-5. MultigetSliceExample.java
   
2. package com.cassandraguide.rw;
   
3. 
   
4. //imports omitted
   
5. 
   
6. public class MultigetSliceExample {
   
7.        
   
8.         private static final ConsistencyLevel CL = ConsistencyLevel.ONE;
   
9.        
   
10.         private static final String columnFamily = "Standard1";
    
11. 
    
12.         public static void main(String[] args) throws UnsupportedEncodingException,
    
13.                         InvalidRequestException, UnavailableException, TimedOutException,
    
14.                         TException, NotFoundException {
    
15. 
    
16.                 Connector conn = new Connector();
    
17.                 Cassandra.Client client = conn.connect();
    
18.                
    
19.                 System.out.println("Running Multiget Slice.");
    
20. 
    
21.                 SlicePredicate predicate = new SlicePredicate();
    
22.                 List<byte[]> colNames = new ArrayList<byte[]>();
    
23.                 colNames.add("a".getBytes());
    
24.                 colNames.add("c".getBytes());
    
25.                 predicate.column_names = colNames;
    
26. 
    
27.                 ColumnParent parent = new ColumnParent(columnFamily);
    
28. 
    
29.                 //instead of one row key, we specify many
    
30.                 List<byte[]> rowKeys = new ArrayList<byte[]>();
    
31.                 rowKeys.add("k1".getBytes());
    
32.                 rowKeys.add("k2".getBytes());
    
33.                
    
34.                 //instead of a simple list, we get a map, where the keys are row keys
    
35.                 //and the values the list of columns returned for each
    
36.                 Map<byte[],List<ColumnOrSuperColumn>> results =
    
37.                         client.multiget_slice(rowKeys, parent, predicate, CL);
    
38.                
    
39.                 for (byte[] key : results.keySet()) {       
    
40.                         List<ColumnOrSuperColumn> row = results.get(key);
    
41.                        
    
42.                         System.out.println("Row " + new String(key) + " --> ");
    
43.                         for (ColumnOrSuperColumn cosc : row) {
    
44.                                 Column c = cosc.column;
    
45.                                 System.out.println(new String(c.name, "UTF-8") + " : "
    
46.                                                 + new String(c.value, "UTF-8"));
    
47.                         }
    
48.                 }
    
49.                                
    
50.                 conn.close();
    
51.                
    
52.                 System.out.println("All done.");
    
53.         }
    
54. }

复制代码

1. Let’s run an example that will delete some data that we previously inserted. Note that there is no “delete” operation in Cassandra, it’s remove, and there’s really no “remove,” it’s just a write (of a tombstone flag). Because a remove operation is really a tombstone write, you still have to supply a timestamp with the operation, because if there are multiple clients writing, the highest timestamp wins—and those writes might include a tombstone or a new value. Cassandra doesn’t discriminate here; whichever operation has the highest timestamp will win.
   
2. 
   
3. A simple delete looks like this:
   
4. 
   
5. Connector conn = new Connector();
   
6. Cassandra.Client client = conn.connect();
   
7. 
   
8. String columnFamily = "Standard1";
   
9. byte[] key = "k2".getBytes(); //this is the row key
   
10. 
    
11. Clock clock = new Clock(System.currentTimeMillis());
    
12. 
    
13. ColumnPath colPath = new ColumnPath();
    
14. colPath.column_family = columnFamily;
    
15. colPath.column = "b".getBytes();
    
16. 
    
17. client.remove(key, colPath, clock, ConsistencyLevel.ALL);
    
18. 
    
19. System.out.println("Remove done.");
    
20. 
    
21. conn.close();

复制代码

1. Batch Mutates
   
2. 
   
3. There were many examples of using batch mutate to perform multiple inserts in Chapter 4, so I won’t rehash that here. I’ll just present an overview.
   
4. 
   
5. To perform many insert or update operations at once, use the batch_mutate method instead of the insert method. Like a batch update in the relational world, the batch_mutate operation allows grouping calls on many keys into a single call in order to save on the cost of network round trips. If batch_mutate fails in the middle of its list of mutations, there will be no rollback, so any updates that have already occured up to this point will remain intact. In the case of such a failure, the client can retry the batch_mutate operation.

复制代码

1. Running the Word Count Example
   
2. 
   
3. Word count is one of the examples given in the MapReduce paper and is the starting point for many who are new to the framework. It takes a body of text and counts the occurrences of each distinct word. Here we provide some code to perform a word count over data contained in Cassandra. A working example of word count is also included in the Cassandra source download.
   
4. 
   
5. First we need a Mapper class, shown in Example 12-1.
   
6. 
   
7. Example 12-1. The TokenizerMapper.java class
   
8. public static class TokenizerMapper extends Mapper<byte[],
   
9.     SortedMap<byte[], IColumn>, Text, IntWritable> {
   
10. 
    
11.   private final static IntWritable one = new IntWritable(1);
    
12.   private Text word = new Text();
    
13.   private String columnName;
    
14. 
    
15.   public void map(byte[] key, SortedMap<byte[], IColumn> columns, Context context)
    
16.     throws IOException, InterruptedException {
    
17. 
    
18.     IColumn column = columns.get(columnName.getBytes());
    
19.     String value = new String(column.value());
    
20.     StringTokenizer itr = new StringTokenizer(value);
    
21. 
    
22.     while (itr.hasMoreTokens()) {
    
23.       word.set(itr.nextToken());
    
24.       context.write(word, one);
    
25.     }
    
26.   }
    
27. 
    
28.   protected void setup(Context context)
    
29.     throws IOException, InterruptedException {
    
30. 
    
31.     this.columnName = context.getConfiguration().get(“column_name”);
    
32. }
    
33. }

复制代码

1. Cassandra Hadoop Source Package
   
2. 
   
3. Cassandra has a Java source package for Hadoop integration code, called org.apache.cassandra.hadoop. There we find:
   
4. 
   
5. ColumnFamilyInputFormat
   
6. The main class we’ll use to interact with data stored in Cassandra from Hadoop. It’s an extension of Hadoop’s InputFormat abstract class.
   
7. 
   
8. ConfigHelper
   
9. A helper class to configure Cassandra-specific information such as the server node to point to, the port, and information specific to your MapReduce job.
   
10. 
    
11. ColumnFamilySplit
    
12. The extension of Hadoop’s InputSplit abstract class that creates splits over our Cassandra data. It also provides Hadoop with the location of the data, so that it may prefer running tasks on nodes where the data is stored.
    
13. 
    
14. ColumnFamilyRecordReader
    
15. The layer at which individual records from Cassandra are read. It’s an extension of Hadoop’s RecordReader abstract class.
    
16. 
    
17. There are similar classes for outputting data to Cassandra in the Hadoop package, but at the time of this writing, those classes are still being finalized.

复制代码