support cassandra

pranab · Mar 3, 2012 · 2902c3d · 2902c3d
1 parent c4f2a25
commit 2902c3d
Show file tree

Hide file tree

Showing 5 changed files with 17 additions and 8 deletions.
diff --git a/src/main/java/org/chargeur/config/SpreadSheetMapper.java b/src/main/java/org/chargeur/config/SpreadSheetMapper.java
@@ -26,6 +26,7 @@
 public class SpreadSheetMapper {
 	private String name;
 	private String database;
+	private String dbConfigFile;
 	private String table;
 	private boolean colNameInFirstRow;
 	private List<String> rowKey;
@@ -57,6 +58,14 @@ public String getDatabase() {
 	public void setDatabase(String database) {
 		this.database = database;
 	}
+	public String getDbConfigFile() {
+		return dbConfigFile;
+	}
+
+	public void setDbConfigFile(String dbConfigFile) {
+		this.dbConfigFile = dbConfigFile;
+	}
+
 	public String getTable() {
 		return table;
 	}

diff --git a/src/main/java/org/chargeur/core/CassandraLoader.java b/src/main/java/org/chargeur/core/CassandraLoader.java
@@ -26,7 +26,6 @@
 import org.apache.cassandra.thrift.ConsistencyLevel;
 
 import agiato.cassandra.data.BatchLoader;
-import agiato.cassandra.data.DataAccess;
 import agiato.cassandra.data.DataManager;
 import agiato.cassandra.data.SuperColumnValue;
 import agiato.cassandra.data.SuperRow;
@@ -49,6 +48,7 @@ public void load(List<ColumnValue> rowKeyValues, List<ColumnValue> columns)
 		 Map<String, SuperColumnValue> superColsMap = new HashMap<String, SuperColumnValue>();
 
 		 for (ColumnValue colVal : columns ) {
+			 //super col name and value
 			 String supCol  = colVal.getColFamily();
 			 SuperColumnValue supColVal = superColsMap.get(supCol);
 			 if (null == supColVal) {
@@ -63,11 +63,12 @@ public void load(List<ColumnValue> rowKeyValues, List<ColumnValue> columns)
 			 supColVal.addValue(caColVal);
 		 }
 
+		 //create super col list and row
 		 for (String name : superColsMap.keySet()) {
 			 superColValues.add(superColsMap.get(name));
 		 }
-
 		SuperRow superRow = new  SuperRow(rowKey, superColValues);
+
 		batchLoader.addRow(rowKey, superColValues);
 	}
 

diff --git a/src/main/java/org/chargeur/core/DbLoader.java b/src/main/java/org/chargeur/core/DbLoader.java
@@ -40,10 +40,9 @@ protected   byte[] getCompositeKeyBytes(List<ColumnValue> rowKeyValues, boolean
 		rowKey = new byte[size];
 		int tgtOffset = 0;
 		for (ColumnValue colVal : rowKeyValues){
-			int srcLength = colVal.getMaxSize();
+			int srcLength = colVal.getSize(paddedString);
 			byte[] srcBytes = paddedString ? colVal.getValueBytesMax() : colVal.getValueBytes();
-			Bytes.putBytes(rowKey, tgtOffset, srcBytes, 0, srcLength);
-			tgtOffset += srcLength;
+			tgtOffset = Bytes.putBytes(rowKey, tgtOffset, srcBytes, 0, srcLength);
 		}
 
 		return rowKey;

diff --git a/src/main/java/org/chargeur/core/Loader.java b/src/main/java/org/chargeur/core/Loader.java
@@ -119,8 +119,6 @@ public void load(String csvFile, String mappingFile, Map<String, String> sideDat
 					}
 				}
 
-
-
 				//load this row
 	            ++count;
 				rowKeyValues = getRowKeyItems(rowComponents);
@@ -154,7 +152,9 @@ private DbLoader createDbLoader() throws Exception {
 
 		if (configurator.getDatabase().equals("hbase")){
 			dbLoader = new HbaseLoader(configurator.getTable(), configurator.getBatchSize());
-		}
+		} else if (configurator.getDatabase().equals("cassandra")){
+			dbLoader = new CassandraLoader(configurator.getDbConfigFile(), configurator.getTable(), configurator.getBatchSize());
+		} 
 
 		return dbLoader;
 	}

diff --git a/target/chargeur-1.0.jar b/target/chargeur-1.0.jar