datadb/compbio/cassandra/CassandraNativeConnector.java

   1 package compbio.cassandra;
   2
   3 import java.util.Calendar;
   4
   5 import org.apache.log4j.Logger;
   6
   7 import com.datastax.driver.core.Cluster;
   8 import com.datastax.driver.core.Host;
   9 import com.datastax.driver.core.Metadata;
  10 import com.datastax.driver.core.ResultSet;
  11 import com.datastax.driver.core.Row;
  12
  13 import com.datastax.driver.core.Session;
  14 import compbio.engine.ProteoCachePropertyHelperManager;
  15 import compbio.util.PropertyHelper;
  16
  17 public class CassandraNativeConnector {
  18         private static Cluster cluster;
  19         private static Session session;
  20         private static final PropertyHelper ph = ProteoCachePropertyHelperManager.getPropertyHelper();
  21         private static Logger log = Logger.getLogger(CassandraNativeConnector.class);
  22
  23         public static String CASSANDRA_HOSTNAME = "localhost";
  24
  25         public static Session getSession () {
  26                 return session;
  27         }
  28
  29         /*
  30          * connect to the cluster and look whether all tables exist
  31          */
  32         public void Connect() {
  33
  34                 String cassandrahostname = ph.getProperty("cassandra.host");
  35                 if (null != cassandrahostname) {
  36                         CASSANDRA_HOSTNAME = cassandrahostname;
  37                 }
  38
  39                 cluster = Cluster.builder().addContactPoint(CASSANDRA_HOSTNAME).build();
  40
  41                 Metadata metadata = cluster.getMetadata();
  42                 System.out.printf("Connected to cluster: %s\n", metadata.getClusterName());
  43                 for (Host host : metadata.getAllHosts()) {
  44                         System.out.printf("Datatacenter: %s; Host: %s; Rack: %s\n", host.getDatacenter(), host.getAddress(), host.getRack());
  45                 }
  46                 session = cluster.connect();
  47                 CreateMainTables();
  48                 System.out.println("Cassandra connected");
  49         }
  50
  51         private void CreateMainTables() {
  52                 session.execute("CREATE KEYSPACE IF NOT EXISTS ProteinKeyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':3};");
  53                 session.execute("USE ProteinKeyspace");
  54
  55                 session.execute("CREATE TABLE IF NOT EXISTS MainParameters "
  56                                 + "(Name ascii, Value ascii, PRIMARY KEY(Name));");
  57
  58                 session.execute("CREATE TABLE IF NOT EXISTS ProteinRow "
  59                                 + "(Protein ascii, JobID ascii, Predictions map<ascii,ascii>, PRIMARY KEY(JobID));");
  60
  61                 session.execute("CREATE TABLE IF NOT EXISTS ProteinLog "
  62                                 + "(JobID ascii, DataBegin ascii, DataEnd ascii, ip ascii, FinalStatus ascii, "
  63                                 + "ExecutionStatus ascii, Protein ascii, PRIMARY KEY(JobID));");
  64
  65                 session.execute("CREATE TABLE IF NOT EXISTS ProteinData "
  66                                 + "(jobtime bigint, JobID ascii, ExecTime int, Protein ascii, PRIMARY KEY(jobtime, JobID));");
  67
  68                 session.execute("CREATE TABLE IF NOT EXISTS FailLog "
  69                                 + "(jobtime bigint, JobID ascii, ExecTime int, ip ascii, FinalStatus ascii, PRIMARY KEY(jobtime, JobID));");
  70
  71                 session.execute("CREATE TABLE IF NOT EXISTS JpredArchive "
  72                                 + "(JobID ascii, Protein varchar, IP ascii, StartTime bigint, ExecTime int, FinalStatus ascii, ExecutionStatus ascii, alignment map<ascii,ascii>, "
  73                                 + "predictions map<ascii,ascii>, ArchiveLink varchar, LOG varchar, PRIMARY KEY(JobID));");
  74
  75                 session.execute("CREATE TABLE IF NOT EXISTS JobDateInfo "
  76                                 + "(jobday bigint, Total bigint,  TotalOK bigint, TotalStopped bigint, TotalError bigint, TotalTimeOut bigint, Program varchar, Version varchar, PRIMARY KEY(jobday));");
  77
  78                 session.execute("CREATE INDEX IF NOT EXISTS ProteinSeq ON ProteinRow (protein);");
  79                 session.execute("CREATE INDEX IF NOT EXISTS ProteinIp ON ProteinLog (ip);");
  80         //      session.execute("CREATE INDEX IF NOT EXISTS JobDateStamp ON ProteinData (jobtime);");
  81         }
  82
  83         public void Closing() {
  84                 session.shutdown();
  85                 cluster.shutdown();
  86                 System.out.println("Cassandra has been shut down");
  87         }
  88
  89         /*
  90          * getting earlest date of jobs from the db
  91          */
  92         public static long getEarliestDateInDB() {
  93                 String com = "SELECT * FROM MainParameters WHERE Name = 'EarliestJobDate';";
  94                 System.out.println("Command: " + com);
  95                 ResultSet results = session.execute(com);
  96
  97                 if (!results.isExhausted()) {
  98                         Row r = results.one();
  99                         return Long.parseLong(r.getString("Value"));
 100                 }
 101                 Calendar cal = Calendar.getInstance();
 102                 return cal.getTimeInMillis();
 103         }
 104
 105 }