1 package compbio.cassandra;
3 import java.io.IOException;
6 import com.datastax.driver.core.Cluster;
7 import com.datastax.driver.core.Host;
8 import com.datastax.driver.core.Metadata;
9 import com.datastax.driver.core.Session;
10 import com.datastax.driver.core.ResultSet;
12 public class CassandraNativeConnector {
13 private static Cluster cluster;
14 private static Session session;
17 * private static Keyspace ksp; private static Mutator<Long> mutatorLong;
18 * private static Mutator<String> mutatorString; private static
19 * Mutator<String> mutatorLog; StringSerializer ss = StringSerializer.get();
20 * LongSerializer ls = LongSerializer.get();
24 * connect to the cluster and look weather the dababase has any data inside
26 public void Connect() {
27 // local cassandra cluster
28 cluster = Cluster.builder().addContactPoint("localhost").build();
29 // distributed cassandra cluster
30 /* cluster = Cluster.builder().addContactPoint("10.0.115.190").build(); */
31 Metadata metadata = cluster.getMetadata();
32 System.out.printf("Connected to cluster: %s\n", metadata.getClusterName());
33 for (Host host : metadata.getAllHosts()) {
34 System.out.printf("Datatacenter: %s; Host: %s; Rack: %s\n", host.getDatacenter(), host.getAddress(), host.getRack());
37 session = cluster.connect();
38 session.execute("CREATE KEYSPACE IF NOT EXISTS ProteinKeyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':3};");
39 session.execute("CREATE COLUMNFAMILY IF NOT EXISTS ProteinKeyspace.ProteinRow (Protein ascii, JobID ascii, Predictions map<ascii,ascii>, PRIMARY KEY(JobID));");
40 session.execute("CREATE COLUMNFAMILY IF NOT EXISTS ProteinKeyspace.ProteinLog "
41 + "(JobID ascii, DataBegin ascii, DataEnd ascii, ip ascii, FinalStatus ascii, ExecutionStatus ascii, Protein ascii, PRIMARY KEY(JobID));");
42 session.execute("CREATE COLUMNFAMILY IF NOT EXISTS ProteinKeyspace.ProteinData (jobtime bigint, JobID ascii, Protein ascii, PRIMARY KEY(JobID));");
44 session.execute("CREATE INDEX ProteinSeq ON ProteinKeyspace.ProteinRow (protein);");
45 session.execute("CREATE INDEX JobDateStamp ON ProteinKeyspace.ProteinData (jobtime);");
47 System.out.println("Cassandra connected");
51 * parsing data source and filling the database
53 public void Parsing() throws IOException {
55 // if (source.equals("http")) {
56 // get data from real Jpred production server
57 System.out.println("Parsing web data source......");
58 String datasrc = "http://www.compbio.dundee.ac.uk/www-jpred/results/usage-new/alljobs.dat";
59 String prefix = "http://www.compbio.dundee.ac.uk/www-jpred/results";
60 JpredParserHTTP parser = new JpredParserHTTP(prefix);
61 parser.Parsing(datasrc, 4);
64 // if (source.equals("file")) {
65 // get irtifical data generated for the DB stress tests
66 System.out.println("Parsing local file data source......");
67 String datasrc = "/home/asherstnev/Projects/Java.projects/proteocache/data_stress_test/data.dat";
68 String prefix = "/home/asherstnev/Projects/Java.projects/proteocache/data_stress_test/Jpreddata";
69 JpredParserLocalFile parser = new JpredParserLocalFile(prefix);
70 parser.Parsing(datasrc, 190);
74 public void Closing() {
76 System.out.println("Cassandra has been shut down");
80 * prepare data for insertion into the db
82 public void InsertData(long jobtime, String startdate, String enddate, String ip, String jobid, String statusEx, String statusFinal,
83 String protein, List<FastaSequence> predictions) {
85 String check1 = "SELECT * FROM ProteinKeyspace.ProteinLog WHERE JobID = '" + jobid + "';";
86 //System.out.println(check1);
87 ResultSet results1 = session.execute(check1);
88 if (results1.isExhausted()) {
89 String com1 = "INSERT INTO ProteinKeyspace.ProteinLog "
90 + "(JobID, IP, DataBegin, DataEnd, FinalStatus, ExecutionStatus, Protein)" + " VALUES ('" + jobid + "','" + ip + "','"
91 + startdate + "','" + enddate + "','" + statusFinal + "','" + statusEx + "','" + protein + "');";
92 //System.out.println(com1);
93 session.execute(com1);
95 String com2 = "INSERT INTO ProteinKeyspace.ProteinData " + "(jobtime, JobID, Protein)" + " VALUES (" + jobtime + ",'" + jobid
96 + "','" + protein + "');";
97 //System.out.println(com2);
98 session.execute(com2);
100 String allpredictions = "";
101 for (FastaSequence pred : predictions) {
102 String predictionname = pred.getId();
103 String prediction = pred.getSequence().replaceAll("\n", "");
104 allpredictions += "'" + predictionname + "':'" + prediction + "',";
106 String final_prediction = "";
107 if (null != allpredictions) {
108 final_prediction = allpredictions.substring(0, allpredictions.length() - 1);
111 String check2 = "SELECT * FROM ProteinKeyspace.ProteinRow WHERE JobID = '" + jobid + "';";
112 //System.out.println(check2);
113 ResultSet results2 = session.execute(check2);
114 if (results2.isExhausted()) {
115 String com3 = "INSERT INTO ProteinKeyspace.ProteinRow " + "(Protein, JobID, Predictions)" + " VALUES ('"
116 + protein + "','" + jobid + "',{" + final_prediction + "});";
117 //System.out.println(com3);
118 session.execute(com3);